AliECS core Kafka producer #520

teo · 2024-02-29T14:45:34Z

No description provided.

knopers8 · 2024-03-01T13:37:40Z

core/protos/o2control.proto

@@ -28,53 +28,22 @@ package o2control;
 option java_package = "ch.cern.alice.o2.control.rpcserver";
 option go_package = "github.com/AliceO2Group/Control/core/protos;pb";

-//////////////// Common event messages ///////////////


If the go files for this proto are stored in coconut/protos and core/protos, why not move the .proto and pb.go files to common/protos?

We should probably do this, yes.

knopers8 · 2024-03-01T14:08:33Z

core/environment/environment.go

+				// We might leave RUNNING not only through STOP_ACTIVITY. In such cases we also need a run stop time.
+				if e.Src == "RUNNING" {
+					endTime, ok := env.workflow.GetUserVars().Get("run_end_time_ms")
+					if ok && endTime == "" {
+						runEndTime := strconv.FormatInt(time.Now().UnixMilli(), 10)
+						env.workflow.SetRuntimeVar("run_end_time_ms", runEndTime)
+					} else {
+						log.WithField("partition", envId.String()).
+							Debug("O2 End time already set before leave_RUNNING")
+					}
+				}


It's not really a part of adding the core Kafka producer. I would propose to remove it from this PR and discuss it separately. We had discussions with Ruben and Chiara which we did not finalize, while this affects the order of EOR and EOX timestamps. I will raise to Vasco that we should bring that discussion to an end.

Fair enough, but the events do need to be emitted at some point.

knopers8 · 2024-03-01T14:22:39Z

core/environment/manager.go

 	// Before we load the workflow, we get the list of currently active detectors. This query must be performed before
 	// loading the workflow in order to compare the currently used detectors with the detectors required by the newly
 	// created environment.
 	alreadyActiveDetectors := envs.GetActiveDetectors()

+	the.EventWriterWithTopic(topic.Environment).WriteEvent(&evpb.Ev_EnvironmentEvent{
+		EnvironmentId:  newId.String(),
+		State:          "PENDING",


I am wondering if this does not mean introducing a new (quasi-)state... At least it could be treated as such by consumers of this topic.

If we do not want to talk about states before the environment exists, perhaps it should be left empty? More of a thought rather than a request.

It kind of is a fake state. It should be fine as far as consumers are concerned, the GUI would likely appreciate it.

Well, we can check to be sure. @graduta @martinboulais ECS will send some events before an environment is considered created and has a state. Do you have any preference whether in such case the "state" field should be empty or contain "PENDING"?

core/task/task.go

teo · 2024-03-21T16:31:22Z

@justonedev1 @knopers8 this was tested on STAGING, the event logs are attached to OCTRL-765
The patched core is still there if you'd like to take it for a spin. I recommend https://github.com/sevagh/pq to consume the topics, details in the Makefile.

justonedev1

I don't see anything wrong with the code itself. However I left some comments regarding code quality and some questions regarding content of this PR, as it seemed, that there as quite a few things bloating this PR while not being necessary and should be done as different PR.

Code quality comments can be taken for the whole code base: there are a lot of messages created in place instead of one parametrized function. This causes really big bloat in function itself masking the logic.

But I don't see the reason why to delay this PR because of these "cosmetic" things.

However, it would be probably best if @knopers8 also takes a look, as he knows much more about functionality and code in general

justonedev1 · 2024-03-21T16:53:49Z

core/environment/environment.go

+			Message:          "transition completed successfully",
+			Transition:       t.eventName(),
+		})
+	}
 	return


Just a comment: I would create some function which creates all these event messages, because right now there are a lot of new lines just declaring new EnvironmentEvents and 'hiding' the actual logic + bloating the length of the function by hundreds of lines of code.

I'm not sure what you mean. We do need to define the environment events we're about to send, and while some fields are the same others change from one event to the other. Whether the fields are arguments to a function or fields of a struct doesn't change the fact that we need to provide them. Maybe I'm missing what you're proposing, can you elaborate?

I am not proposing changes of functionality, I just see that you are creating a lot of pb.Ev_EnvironmentEvent messages in place with a lot of same parameters, the only changes are for parameters Message and Error ... So it would be great to create a function (eg. createEV_Environment) that would do it. and you would call it just like this: createEv_Environment(env, error, message). This would reduce the bloat of the function significantly.

justonedev1 · 2024-03-21T17:05:27Z

core/integration/odc/plugin.go

@@ -1357,128 +1377,6 @@ func (p *Plugin) CallStack(data interface{}) (stack map[string]interface{}) {
 		}
 		return
 	}
-	stack["ConfigureLegacy"] = func() (out string) {


Do I guess correctly, that this has nothing to do with kafka producer?

No, this is a bit of legacy code we haven't been using for over a year, perhaps years. I remove it because we don't need it any more, and I do it as part of this PR because it saves me the trouble of implementing event production for this function.

OK, I still think that it should have been separate branch and PR before of this PR, but it is just nitpicking

justonedev1 · 2024-03-21T17:23:36Z

core/serverutil.go

-		streams: make(map[string]chan *pb.Event),
-	}
-}
-


If I understood correctly, you are just moving this somewhere else... is it connected to kafka producer?

Yes, streamutil.go. It's part of a general cleanup at the very start of this effort.

knopers8

Thanks, @teo !

I agree with @justonedev1 regarding reducing the boiler-plate where possible, because indeed these calls are quite verbose and make reading the code more inconvenient. I don't have any clear ideas to propose though, so I leave it up to you to consider if something can be done about it.

Then, since this the kafka events might be accessed by multiple parties with different familiarity in the repository structure, it would be extremely useful to have some documentation of what is published, where to find proto files and what are possible values of less obvious fields like Ev_TaskEvent.className, Ev_TaskEvents.path, Ev_CallEvent.path. For plugins, the expected JSON payloads should be documented. I suppose you could modify the existing kafka.md, which will become obsoleted.

I don't have much detailed comments, perhaps I missed something, but the PR is really huge. I believe it could have been done in smaller chunks, but this is only my suggestion for the future.

common/protos/events.proto

common/event/topic/topic.go

teo · 2024-03-22T10:57:55Z

Thanks, @teo !

I agree with @justonedev1 regarding reducing the boiler-plate where possible, because indeed these calls are quite verbose and make reading the code more inconvenient. I don't have any clear ideas to propose though, so I leave it up to you to consider if something can be done about it.

Then, since this the kafka events might be accessed by multiple parties with different familiarity in the repository structure, it would be extremely useful to have some documentation of what is published, where to find proto files and what are possible values of less obvious fields like Ev_TaskEvent.className, Ev_TaskEvents.path, Ev_CallEvent.path. For plugins, the expected JSON payloads should be documented. I suppose you could modify the existing kafka.md, which will become obsoleted.

I don't have much detailed comments, perhaps I missed something, but the PR is really huge. I believe it could have been done in smaller chunks, but this is only my suggestion for the future.

Thank you both for reviewing. First the easy part: I've added some comments to the protofile to clarify these fields, and I've included some documentation in kafka.md as proposed.

As for the verbosity of the event instantiations, I understand they can get repetitive and I see your points. Generally in Go one values simplicity and sometimes verbosity over hiding things within constructors, factories and such, but repetition repetition can get annoying annoying. I'll give this a serious thought for the next iteration, in the meantime I propose we merge as it is.

teo force-pushed the kafka-producer branch from 0c2bbb6 to 1309105 Compare February 29, 2024 14:47

teo marked this pull request as draft February 29, 2024 14:48

teo requested review from knopers8 and justonedev1 February 29, 2024 14:48

knopers8 reviewed Mar 4, 2024

View reviewed changes

teo force-pushed the kafka-producer branch 2 times, most recently from 6ea2809 to 4cec802 Compare March 14, 2024 15:01

teo added 23 commits March 19, 2024 14:14

[core] Refactor serverutil in preparation for eventstream

f963751

[core] Events protofile

25fdadd

[core] Additional events

ca1c759

[core] Update events.proto+o2control.proto with NewEnvironmentAsync

0f92acb

[common] Use events.proto in o2control.proto

cf40d75

[coconut] Fix Protobuf generator call

8c5f41f

[core] Kafka wrapper

34b7a0d

[core] Emit environment events

0a7c61b

[coconut] Add asynchronous mode (-y) to coconut env create command

e983b2c

[coconut] Implement coconut env create -y

fabece2

[core] Add task traits and CallEvent to events.proto

d13d12b

[common] Add CallEvent

231d3d9

[core] Emit call events to inform on plugin calls

aa4ac3f

[core] Send EnvId with TaskEvents

95c41fe

[core] Rename busEvent in task.go

225a5f9

[core] Add IntegratedServiceEvent and rename Envid field

48499fb

[core] Push env vars on workflow load

fe2dc90

[common] Allow event creation with specific timestamp

7530576

[common] Various additions to events in events.proto

28c3e68

[build] Bump dependencies

6b7c8fd

[core] Include parent role path in task events

92d513f

[core] Improve call information in CallEvents

e82a40e

[core] Emit IntegratedServiceEvents from DCS

dc190db

teo added 10 commits March 19, 2024 14:17

[core] Make sure we always output ECS detector codes, not DCS ones

8c57a45

[core] Don't forget to include error in DCS ERROR events

274e73f

[core] Better DCS event descriptions

b3e3461

[core] Emit ddscheduler events

27eec71

[core] Remove legacy ODC handlers

68b2c46

[core] Emit ODC events

16ce849

[core] Emit TRG events

db62a9f

[common] Enable AllowAutoTopicCreation in Kafka client

ae22193

[core] Correct Kafka topic

831178d

[build] Generate fdset file for decoding Kafka messages with pq

9c62431

teo force-pushed the kafka-producer branch from 57080db to 9c62431 Compare March 19, 2024 13:20

teo added 7 commits March 19, 2024 17:05

[core] Emit call events to aliecs.call topic and include envId

d17a4a1

[core] Enable IntegratedServiceEvents

8695b2a

[core] Pass IntegratedServiceEvents by ref

31eb36d

[core] Write to Kafka asynchronously

e733190

[core] Nullify odc Devices list before emitting events

6ffe6b6

[core] Trim down ODC events some more

d8785f0

[core] Publish ODC partition state changes

e3a392f

teo marked this pull request as ready for review March 21, 2024 16:28

justonedev1 previously approved these changes Mar 21, 2024

View reviewed changes

knopers8 reviewed Mar 22, 2024

View reviewed changes

common/protos/events.proto Outdated Show resolved Hide resolved

common/event/topic/topic.go Outdated Show resolved Hide resolved

[core] Document events.proto and change currentRunNumber field

e3d59c1

teo dismissed justonedev1’s stale review via e3d59c1 March 22, 2024 10:31

teo added 2 commits March 22, 2024 11:32

[core] Document currently unused topics

62dcce8

[docs] Document Kafka producer functionality

f7c2786

knopers8 approved these changes Mar 22, 2024

View reviewed changes

teo merged commit 8ec6061 into master Mar 22, 2024
2 checks passed

teo deleted the kafka-producer branch March 22, 2024 11:21

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

AliECS core Kafka producer #520

AliECS core Kafka producer #520

teo commented Feb 29, 2024

knopers8 Mar 1, 2024

teo Mar 8, 2024

knopers8 Mar 1, 2024

teo Mar 8, 2024

knopers8 Mar 1, 2024

teo Mar 8, 2024

knopers8 Mar 15, 2024

teo commented Mar 21, 2024 •

edited

Loading

justonedev1 left a comment •

edited

Loading

justonedev1 Mar 21, 2024

teo Mar 22, 2024

justonedev1 Mar 22, 2024

justonedev1 Mar 21, 2024

teo Mar 22, 2024

justonedev1 Mar 22, 2024

justonedev1 Mar 21, 2024

teo Mar 22, 2024

justonedev1 Mar 22, 2024

knopers8 left a comment

teo commented Mar 22, 2024

AliECS core Kafka producer #520

AliECS core Kafka producer #520

Conversation

teo commented Feb 29, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

teo commented Mar 21, 2024 • edited Loading

justonedev1 left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

knopers8 left a comment

Choose a reason for hiding this comment

teo commented Mar 22, 2024

teo commented Mar 21, 2024 •

edited

Loading

justonedev1 left a comment •

edited

Loading