Vector panics with "Adding duplicate output id to fanout" when config reloads

[khaskelberg]

A note for the community

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Problem

When you remove one of the route inputs from sinks (or transforms), configuration reload and logs stop flowing when this input is reached.
After you add the input again - vector crashes with "Adding duplicate output id to fanout".

We expect that vector continues working as expected after the reload

Configuration

Original Configuration

data_dir = "/var/lib/vector"

# Input data. Change me to a valid input source.
[sources.in]
type = "demo_logs"
format = "syslog"

[transforms.parser]
type = "remap"
inputs = ["in"]
source = """
. |= parse_syslog!(.message)
cur = now()
rand = slice!(to_string(to_unix_timestamp(cur)), -1) # just a way to generate random number from 0..9
.id = to_int!(rand)
"""

[transforms.id_router]
type="route"
inputs=["parser"]
route.id_0 = ".id == 0"
route.id_1 = ".id == 1"
route.id_2 = ".id == 2"
route.id_3 = ".id == 3"
route.id_4 = ".id == 4"
route.id_5 = ".id == 5"
route.id_6 = ".id == 6"
route.id_7 = ".id == 7"
route.id_8 = ".id == 8"
route.id_9 = ".id == 9"

# Output data
[sinks.console]
inputs = ["id_router.id_0", "id_router.id_1", "id_router.id_2", "id_router.id_3", "id_router.id_4", "id_router.id_5", "id_router.id_6", "id_router.id_7", "id_router.id_8", "id_router.id_9", "id_router._unmatched"]
type = "console"
target = "stdout"
encoding = "json"

[sinks.console_1]
inputs = ["id_router.id_0"]
type = "console"
target = "stdout"
encoding = "json"

First config change

data_dir = "/var/lib/vector"

# Input data. Change me to a valid input source.
[sources.in]
type = "demo_logs"
format = "syslog"

[transforms.parser]
type = "remap"
inputs = ["in"]
source = """
. |= parse_syslog!(.message)
cur = now()
rand = slice!(to_string(to_unix_timestamp(cur)), -1) # just a way to generate random number from 0..9
.id = to_int!(rand)
"""

[transforms.id_router]
type="route"
inputs=["parser"]
route.id_0 = ".id == 0"
route.id_1 = ".id == 1"
route.id_2 = ".id == 2"
route.id_3 = ".id == 3"
route.id_4 = ".id == 4"
route.id_5 = ".id == 5"
route.id_6 = ".id == 6"
route.id_7 = ".id == 7"
route.id_8 = ".id == 8"
route.id_9 = ".id == 9"

# Output data
[sinks.console]
inputs = ["id_router.id_1", "id_router.id_2", "id_router.id_3", "id_router.id_4", "id_router.id_5", "id_router.id_6", "id_router.id_7", "id_router.id_8", "id_router.id_9", "id_router._unmatched"]
type = "console"
target = "stdout"
encoding = "json"

[sinks.console_1]
inputs = ["id_router.id_0"]
type = "console"
target = "stdout"
encoding = "json"

Second config change

data_dir = "/var/lib/vector"

# Input data. Change me to a valid input source.
[sources.in]
type = "demo_logs"
format = "syslog"

[transforms.parser]
type = "remap"
inputs = ["in"]
source = """
. |= parse_syslog!(.message)
cur = now()
rand = slice!(to_string(to_unix_timestamp(cur)), -1) # just a way to generate random number from 0..9
.id = to_int!(rand)
"""

[transforms.id_router]
type="route"
inputs=["parser"]
route.id_0 = ".id == 0"
route.id_1 = ".id == 1"
route.id_2 = ".id == 2"
route.id_3 = ".id == 3"
route.id_4 = ".id == 4"
route.id_5 = ".id == 5"
route.id_6 = ".id == 6"
route.id_7 = ".id == 7"
route.id_8 = ".id == 8"
route.id_9 = ".id == 9"

# Output data
[sinks.console]
inputs = ["id_router.id_0", "id_router.id_1", "id_router.id_2", "id_router.id_3", "id_router.id_4", "id_router.id_5", "id_router.id_6", "id_router.id_7", "id_router.id_8", "id_router.id_9", "id_router._unmatched"]
type = "console"
target = "stdout"
encoding = "json"

[sinks.console_1]
inputs = ["id_router.id_0"]
type = "console"
target = "stdout"
encoding = "json"

Version

vector 0.22.1 (x86_64-unknown-linux-gnu b633e95 2022-06-10)

Debug Output

https://gist.github.com/khaskelberg/27da768c4a254b1655802d56d519a6fa

In the example above you can find that after the first reload, on line 35 (https://gist.github.com/khaskelberg/27da768c4a254b1655802d56d519a6fa#file-vector-output-log-L35) logs just stop flowing, and after the second reload everything crashes

Example Data

In my example - I first remove id_router.id_0 input from sinks.console, config reloads, then I add it again

Additional Context

References

No response

[khaskelberg]

After futher testing - you can reproduce the crashing with this configuration, if you delete and readd parser in sinks.console while watching the config

data_dir = "/var/lib/vector"

# Input data. Change me to a valid input source.
[sources.in]
type = "demo_logs"
format = "syslog"


[transforms.parser]
type = "remap"
inputs = ["in"]
source = """
. |= parse_syslog!(.message)
"""

[transforms.parser_1]
type = "remap"
inputs = ["in"]
source = """
. |= parse_syslog!(.message)
"""

# Output data
[sinks.console]
inputs = ["parser", "parser_1"]
type = "console"
target = "stdout"
encoding = "json"

[zhongzc]

Fixing

[khaskelberg]

Fixing

Tested with the new PR. Works as expected. Thanks for the quick fix! Great work!