Set unique flow id & timestamp to each flow

[sananguliyev]

In the current build of the traced stream, tracing events are currently grouped only by section. This setup makes it difficult to link these events together and understand the complete message journey in the stream flow. The new implementation addresses this issue by:

• Assigning a unique flow ID to each message journey
• Adding a timestamp to each tracing event to provide a clearer order of the steps the message went through
• Introducing EventsByFlowID, which organizes events by flow and arranges them in chronological order based on timestamp.

Additionally: If no tags exist: git describe --tags will fail, but the error message will be suppressed by 2>/dev/null, and then it will fall back to "v0.0.0". Currently, it prints fatal: No names found, cannot describe anything. when make file tries to generate version.

Fixes #344

[sananguliyev]

@jem-davies @gregfurman
Could you please take a look when you have time?

[gregfurman]

@sananguliyev Hey! Thanks for the contribution and apologies for the delay. Will take a look this eve 😄

[gregfurman]

Some questions relating to the metadata approach. Think better internal tracing is a brilliant idea so I'm keen to hear your thoughts on my comments 😄

[gregfurman]

Thanks for the follow ups on this!

I'm thinking whether we want to ever set the flowID values outside of the traceInput (i.e when a message part is first created), although I suppose there's the case where we want to create child-spans.

bento/internal/bundle/tracing/input.go

Lines 52 to 57 in 852a365

	if t.e.IsEnabled() {
	_ = tran.Payload.Iter(func(i int, part *message.Part) error {
	_ = atomic.AddUint64(t.ctr, 1)
	t.e.Add(EventProduceOf(part))
	return nil
	})

Let me know!

[sananguliyev]

Thanks for the follow ups on this!

I'm thinking whether we want to ever set the flowID values outside of the traceInput (i.e when a message part is first created), although I suppose there's the case where we want to create child-spans.

bento/internal/bundle/tracing/input.go

Lines 52 to 57 in 852a365

	if t.e.IsEnabled() {
	_ = tran.Payload.Iter(func(i int, part *message.Part) error {
	_ = atomic.AddUint64(t.ctr, 1)
	t.e.Add(EventProduceOf(part))
	return nil
	})

Let me know!

Why We Need Both

1. flowIDInput: Ensures flow IDs are always available (even without tracing)
2. tracedInput: Records what happens with those flow IDs (only when user wants tracing)

---

• flowIDInput = "Give every message a passport (flow ID)"
• tracedInput = "Write down in a log book every time a message passes through"

And this also helps to group events not only by section but also by flow.

Please, take another look. Thanks in advance. @gregfurman

[gregfurman]

Thanks for your patience with this @sananguliyev 🙏 I've given it a final look and it's looking great! -- although I think we should maybe not enable this by default (incase of bugs).

I think a BuildTracedV2 approach would work best and we can change over the BuildTraced once we're confident the new flowID is working as intended.

cc @jem-davies since this is kinda a far reaching change

[gregfurman]

nit: why internaltracing alias here?

[sananguliyev]

Without the alias, there would be a naming conflict. The current package is already tracing, so importing another package with the same name would be ambiguous. The alias internaltracing makes it clear which tracing package is being referenced when used in the code.

[gregfurman]

Once this FlowID approach is GA, perhaps we unify this wrapWithFlowID approach with the traceInput approach. Perhaps we add a TODO for that. Thoughts?

[sananguliyev]

Yes, sure. Is writing code comments enough, or do you want to somehow document it somewhere?

[sananguliyev]

@gregfurman I have added new BuildTracedV2 as you suggested, which was good idea 👍

Please, take another look to finalize this implementation.