lose data when sending data in the same time

[clyang82]

Refer to the code in security alert tests:

It("Should be able to sync security alert counts event from multiple central instances in hub", func() {
		By("Create event")
		version1 := eventversion.NewVersion()
		version1.Incr()
		dataEvent1 := &wiremodels.SecurityAlertCounts{
			Low:       1,
			Medium:    2,
			High:      3,
			Critical:  4,
			DetailURL: DetailURL,
			Source:    source1,
		}
		event1 := ToCloudEvent(leafHubName, string(enum.SecurityAlertCountsType), version1, dataEvent1)

		version2 := eventversion.NewVersion()
		version2.Incr()
		dataEvent2 := &wiremodels.SecurityAlertCounts{
			Low:       1,
			Medium:    2,
			High:      3,
			Critical:  4,
			DetailURL: DetailURL,
			Source:    source2,
		}
		event2 := ToCloudEvent(leafHubName, string(enum.SecurityAlertCountsType), version2, dataEvent2)

		By("Sync events with transport")
		err := producer.SendEvent(ctx, *event1)
		Expect(err).To(Succeed())

		err = producer.SendEvent(ctx, *event2)
		Expect(err).To(Succeed())

Send two events in the same time, the event2 is lost sometimes.

/assign @yanmxa

[yanmxa]

Since you use different version variables to control the events of the same type. The global hub manager will ignore the second item(with the same version). You can try to update the code with the following pattern:

It("Should be able to sync security alert counts event from multiple central instances in hub", func() {
		By("Create event")
		version1 := eventversion.NewVersion()
		version1.Incr()
		dataEvent1 := &wiremodels.SecurityAlertCounts{
			Low:       1,
			Medium:    2,
			High:      3,
			Critical:  4,
			DetailURL: DetailURL,
			Source:    source1,
		}
		event1 := ToCloudEvent(leafHubName, string(enum.SecurityAlertCountsType), version1, dataEvent1)

                version1.Incr()
		dataEvent2 := &wiremodels.SecurityAlertCounts{
			Low:       1,
			Medium:    2,
			High:      3,
			Critical:  4,
			DetailURL: DetailURL,
			Source:    source2,
		}
		event2 := ToCloudEvent(leafHubName, string(enum.SecurityAlertCountsType), version1, dataEvent2)

		By("Sync events with transport")
		err := producer.SendEvent(ctx, *event1)
		Expect(err).To(Succeed())

		err = producer.SendEvent(ctx, *event2)
		Expect(err).To(Succeed())

[jhernand]

Since you use different version variables to control the events of the same type. The global hub manager will ignore the second item(with the same version). You can try to update the code with the following pattern:

@yanmxa I tried that locally and I still get the same failure, so I think there is something else.

But independent of this test, should we be increasing the version number with each message that we send? Currently we aren't, we are always using 0.1:

multicluster-global-hub/agent/pkg/status/controller/security/stackrox_syncer.go

Line 434 in 5e0e5c5

version, err := version.VersionFrom("0.1")

Should we change that?

[yanmxa]

@jhernand Yes! each type of message has a version, when the message is modified or sent, the version will increase. So the manager will always handle the message with the new version.

If the manager receives a message with an older version or the same as before. It assumes the message is expired and ignore it.

[jhernand]

@jhernand Yes! each type of message has a version, when the message is modified or sent, the version will increase. So the manager will always handle the message with the new version.

If the manager receives a message with an older version or the same as before. It assumes the message is expired and ignore it.

OK, thanks @yanmxa. @danmanor can you take care of that?

@yanmxa let me insist that I tried what you suggested in the test, and it still fails from time to time if I remove the sleep.

[danmanor]

@yanmxa I am not sure I fully understand the versioning mechanism here. What should be the first version value ? if we set it to 0.0, I don't see how to first message is sent because

func (h *genericEmitter) ShouldSend() bool {
	return h.currentVersion.NewerThan(&h.lastSentVersion)
}

[yanmxa]

Hi @danmanor !

Sorry, I missed something. Let us change the test case to

It("Should be able to sync security alert counts event from multiple central instances in hub", func() {
    By("Create event")
    version := eventversion.NewVersion()
    version.Incr()
    dataEvent1 := &wiremodels.SecurityAlertCounts{
      Low:       1,
      Medium:    2,
      High:      3,
      Critical:  4,
      DetailURL: DetailURL,
      Source:    source1,
    }

    By("Sync event1 with transport")
    event1 := ToCloudEvent(leafHubName, string(enum.SecurityAlertCountsType), version, dataEvent1)
    err := producer.SendEvent(ctx, *event1)
    Expect(err).To(Succeed())
    version.Next()

    version.Incr()
    dataEvent2 := &wiremodels.SecurityAlertCounts{
      Low:       1,
      Medium:    2,
      High:      3,
      Critical:  4,
      DetailURL: DetailURL,
      Source:    source2,
    }
    event2 := ToCloudEvent(leafHubName, string(enum.SecurityAlertCountsType), version, dataEvent2)

    By("Sync event2 with transport")
    err = producer.SendEvent(ctx, *event2)
    Expect(err).To(Succeed())
}

We only send events if the version has changed. I didn't realize that you had sent both event1 and event2 simultaneously earlier. I also sent you a message on Slack—if you're still facing any issues, feel free to reach out, and we can discuss it!

[danmanor]

opened - #1123