enhancement(fluent source): Add support for forwarding over unix socket

[tustvold]

Summary

Change Type

• Bug fix
• New feature
• Non-functional (chore, refactoring, docs)
• Performance

Is this a breaking change?

• Yes
• No

How did you test this PR?

I couldn't work out how to get the integration tests working, but I tested by running fluent-bit with the following config

[SERVICE]
    Flush      5
    Daemon     off
    Log_Level  info

[INPUT]
    Name       cpu
    Tag        cpu_usage

[OUTPUT]
    Name          forward
    Match         *
    unix_path     /tmp/map/fluent.sock

And then running vector with

sources:
  fluent:
    type: fluent
    path: /tmp/map/fluent.sock

sinks:
  out:
    inputs:
      - fluent
    type: console
    encoding:
      codec: json

And seeing the metrics being transferred.

I think it should be a relatively straightforward lift for someone with familiarity with the existing integration testing harness to hook this up, but I got all sorts of errors.

Does this PR include user facing changes?

• Yes. Please add a changelog fragment based on our guidelines.
• No. A maintainer will apply the "no-changelog" label to this PR.

Checklist

• Please read our Vector contributor resources.
  • make check-all is a good command to run locally. This check is
    defined here. Some of these
    checks might not be relevant to your PR. For Rust changes, at the very least you should run:
    • cargo fmt --all
    • cargo clippy --workspace --all-targets -- -D warnings
    • cargo nextest run --workspace (alternatively, you can run cargo test --all)
• If this PR introduces changes Vector dependencies (modifies Cargo.lock), please
  run dd-rust-license-tool write to regenerate the license inventory and commit the changes (if any). More details here.

References

• Closes: Source: fluent, allow sending events to unix socket #17132

[bits-bot]

All committers have signed the CLA.

[tustvold]

Ultimately this change is largely mechanical and is just gluing together the existing decoder with the existing unix stream machinery, but there is definitely the possibility I have missed something subtle.

[tustvold]

Unlike the socket data source I used untagged, to avoid this being a breaking change requiring introducing a "mode" tag. As the fields path vs address are unlikely to ever overlap, I think this is probably fine

[tustvold]

I don't really understand what this does, I just matched the socket source

[tustvold]

This change is necessary to allow using the FluentDecoder. As the fluent protocol doesn't have an explicit framing mechanism that I could ascertain, there wasn't a good way to make this work with the codecs::Decoder setup which separates framing and decoding.

This, and better discoverability, is also the reason I opted not to implement this as a codec on the socket source, as suggested on the ticket

[pront]

Thank you for this contribution @tustvold, I will review this PR in the following days (currently have a big backlog).