[V2][Feat] Add events service by machichima · Pull Request #6970 · flyteorg/flyte

machichima · 2026-03-06T00:09:48Z

Tracking issue

Why are the changes needed?

Previous we call internal run service to record action events (persist to DB) from action client (k8s watcher). However, we cannot get the full info we want from the CR (e.g. LogInfo, ErrorInfo, ...). In this PR, we add a new EventService running independently, receiving requests from executor and forward to internal run service through background workers.

Alternatives

We've considered following alternatives:

1. Store map of `ActionEvents` under `TaskAction` CR status

We can store a map of action events with ID under the TaskAction CR status, and add a persisted_id slice to record the persisted action events. Executor will put new action event to the CR, and action service will take the action event out -> check if it's persisted by checking persisted_id, and call request internal run service.

This is using TaskAction CR as a queue, which we need to:

take the action event info out
determine which action event is persisted
remove the persisted action event

This method has following disadvantages:

can be more complex, as we need to handle the queueing logic on our own
may cause performance issue in large scale
Using CR object as queue is not a good practice

2. Store `ActionEvent` in a new CR

We can add a new ActionEvent CR to store the action events. Executor will create a new ActionEvent CR when there's status update, and action service can watch this CR. Whenever there's new ActionEvent CR create, it will parse info from the CR and send request to internal run service for persisting action event.

This method has following disadvantages:

If run service is down, we may create overwhelming CRs, causing pressure to etcd

What changes were proposed in this pull request?

Add new EventsService proto and implementations
Start a background goroutine EventsWorker to queue and send requests to internal run service (through unary)
Clean-up executor/cmd/main.go and remove unused code

TODO next: Use streaming endpoint / implement batch request from EventsService to internal run service

How was this patch tested?

Start Flyte locally following manager/DEVELOPMENT.md
Use this config in flyte-sdk

admin:
  endpoint: dns:///localhost:8090
  insecure: True
image:
  builder: local
task:
  domain: development
  project: testproject
  org: testorg

Run an example (e.g. python examples/basics/hello_v2.py)
Look into the SQLite DB (flyte.db), check if the data exists

DB:

Labels

Please add one or more of the following labels to categorize your PR:

added: For new features.
changed: For changes in existing functionality.
deprecated: For soon-to-be-removed features.
removed: For features being removed.
fixed: For any bug fixed.
security: In case of vulnerabilities

This is important to improve the readability of release notes.

Setup process

Screenshots

Check all the applicable boxes

I updated the documentation accordingly.
All new and existing tests passed.
All commits are signed-off.

Related PRs

Docs link

main
- Flyte 2 WIP #6583
  - [V2][Feat] Add events service #6970 👈

Signed-off-by: machichima <nary12321@gmail.com>

executor/setup.go

pingsutw · 2026-03-09T02:45:57Z

executor/pkg/controller/taskaction_controller.go

+		Events: []*workflow.ActionEvent{actionEvent},
+	})); err != nil {
+		// Best effort: status update succeeded, do not fail reconciliation on event publish errors.
+		logger.Error(err, "failed to publish action event", "action", actionEvent.GetId().GetName())


What If the events proxy is temporarily down? events are silently lost

I raise an error here in: e1a9f18

And we can rely on controller-runtime's reconcile error backoff logic

Signed-off-by: machichima <nary12321@gmail.com>

pingsutw · 2026-03-09T05:31:08Z

executor/pkg/controller/taskaction_controller.go

 	logger.Info("updateTaskActionStatus", "name", oldTaskAction.Name, "old status", oldTaskAction.Status, "new status", newTaskAction.Status)
-	err := r.Status().Update(ctx, newTaskAction)
-	if err != nil {
+	if err := r.Status().Update(ctx, newTaskAction); err != nil {


We should only update the status after the event has been successfully sent to the event service. Otherwise, the event won’t be resent if it gets lost, since the old and new action statuses would be the same

popojk

LGTM. Left a few small ones

popojk · 2026-03-09T06:04:35Z

executor/pkg/controller/taskaction_controller.go

+	return timestamppb.Now()
+}
+
+func outputRefs(runOutputBase, actionName string) *task.OutputReferences {


Is the actionName always be different if we execute the same task multiple times? Just want to make sure we will not have conflict bucket name.

The root action is created in:

flyte/runs/service/run_service.go

Lines 112 to 121 in e1a9f18

case *workflow.CreateRunRequest_RunId:

org = id.RunId.Org

project = id.RunId.Project

domain = id.RunId.Domain

name = id.RunId.Name

case *workflow.CreateRunRequest_ProjectId:

org = id.ProjectId.Organization

project = id.ProjectId.Name

domain = id.ProjectId.Domain

name = fmt.Sprintf("run-%d", time.Now().Unix())

And for sub-actions, they are created in:
https://github.com/flyteorg/flyte-sdk/blob/a87584a9bc7a2601a6754b4cc3dd2d6481221904/src/flyte/_internal/controllers/remote/_controller.py#L186-L188

I think for sub-actions, we will always have a unique name. For root action, unique name is guaranteed when request with ProjectId. If request with RunId, CreateRun will raise error.

Therefor I think it's safe here

popojk · 2026-03-09T06:45:52Z

executor/pkg/controller/taskaction_controller.go

+			continue
+		}
+		// Return the first explicit cache status signal.
+		if resource.CacheStatus != core.CatalogCacheStatus_CACHE_DISABLED || len(resources) == 1 {


nit: In len(resources) == 1 the number "1" is kind of magical for me. Could we have a comment to explain why we returnresource.CacheStatus when len(resources) == 1?

I was thinking to return the cache status directly if there's only one resource. However, I think this is redundant.

If the only resource is not CatalogCacheStatus_CACHE_DISABLED, it will get into if resource.CacheStatus != core.CatalogCacheStatus_CACHE_DISABLED block, otherwise it will continue to the end of the function and return CatalogCacheStatus_CACHE_DISABLED.

I'll remove it here

Done in 98ad3cc

Signed-off-by: machichima <nary12321@gmail.com>

…ervice Signed-off-by: machichima <nary12321@gmail.com>

Signed-off-by: machichima <nary12321@gmail.com>

machichima · 2026-03-10T01:25:04Z

Got error when merging v2, fixing

Signed-off-by: machichima <nary12321@gmail.com>

machichima added 8 commits March 5, 2026 13:27

feat: add EventsService proto

b65ecc5

Signed-off-by: machichima <nary12321@gmail.com>

feat: add event background worker

5b72469

Signed-off-by: machichima <nary12321@gmail.com>

feat: add events service

053ad12

Signed-off-by: machichima <nary12321@gmail.com>

feat: add to main.go

5742496

Signed-off-by: machichima <nary12321@gmail.com>

refactor: cleanup executor cmd main.go

b8033fd

Signed-off-by: machichima <nary12321@gmail.com>

feat: call eventsClient.Record in executor when having update

5ac18d0

Signed-off-by: machichima <nary12321@gmail.com>

docs: update development guide

10f85d8

Signed-off-by: machichima <nary12321@gmail.com>

feat: add scripts

bdaa5c5

Signed-off-by: machichima <nary12321@gmail.com>

machichima changed the title ~~[V2][Feat] Add event proxy service~~ [V2][Feat] Add events service Mar 6, 2026

github-actions bot mentioned this pull request Mar 6, 2026

Flyte 2 WIP #6583

Draft

3 tasks

machichima mentioned this pull request Mar 6, 2026

[V2] Backgroud goroutine to persist action state #6936

Closed

machichima added 3 commits March 6, 2026 15:19

feat: events record use direct RPC req & remove background worker

7761d4f

Signed-off-by: machichima <nary12321@gmail.com>

refactor: rename to events proxy service proto

d2cf6fe

Signed-off-by: machichima <nary12321@gmail.com>

refactor: rename to events proxy service

a17628b

Signed-off-by: machichima <nary12321@gmail.com>

machichima force-pushed the add-event-proxy-service branch from 81e5562 to a17628b Compare March 9, 2026 01:13

pingsutw reviewed Mar 9, 2026

View reviewed changes

machichima added 2 commits March 9, 2026 10:54

refactor: StateServiceURL to EventsServiceURL

b7cf247

Signed-off-by: machichima <nary12321@gmail.com>

fix: raise error when fail to persist action event

e1a9f18

Signed-off-by: machichima <nary12321@gmail.com>

pingsutw reviewed Mar 9, 2026

View reviewed changes

popojk reviewed Mar 9, 2026

View reviewed changes

fix: remove redundant if condition

98ad3cc

Signed-off-by: machichima <nary12321@gmail.com>

pingsutw added the flyte2 label Mar 9, 2026

pingsutw assigned machichima Mar 9, 2026

machichima added 2 commits March 10, 2026 08:06

fix: update status after persisted

ab5ee80

Signed-off-by: machichima <nary12321@gmail.com>

Merge branch 'v2' of github.com:flyteorg/flyte into add-event-proxy-s…

28abf1c

…ervice Signed-off-by: machichima <nary12321@gmail.com>

pingsutw approved these changes Mar 10, 2026

View reviewed changes

fix: remove unused code

bab1461

Signed-off-by: machichima <nary12321@gmail.com>

fix: bug + remove cluster command

13b38d0

Signed-off-by: machichima <nary12321@gmail.com>

machichima force-pushed the add-event-proxy-service branch from ea15830 to 13b38d0 Compare March 10, 2026 01:29

machichima merged commit af8de97 into v2 Mar 10, 2026
5 checks passed

machichima deleted the add-event-proxy-service branch March 10, 2026 01:37

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[V2][Feat] Add events service#6970

[V2][Feat] Add events service#6970
machichima merged 18 commits intov2from
add-event-proxy-service

machichima commented Mar 6, 2026 •

edited by github-actions bot

Loading

Uh oh!

Uh oh!

pingsutw Mar 9, 2026

Uh oh!

machichima Mar 9, 2026

Uh oh!

pingsutw Mar 9, 2026

Uh oh!

popojk left a comment

Uh oh!

popojk Mar 9, 2026

Uh oh!

machichima Mar 9, 2026

Uh oh!

popojk Mar 9, 2026

Uh oh!

machichima Mar 9, 2026 •

edited

Loading

Uh oh!

machichima Mar 9, 2026

Uh oh!

machichima commented Mar 10, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

	case *workflow.CreateRunRequest_RunId:
	org = id.RunId.Org
	project = id.RunId.Project
	domain = id.RunId.Domain
	name = id.RunId.Name
	case *workflow.CreateRunRequest_ProjectId:
	org = id.ProjectId.Organization
	project = id.ProjectId.Name
	domain = id.ProjectId.Domain
	name = fmt.Sprintf("run-%d", time.Now().Unix())

Conversation

machichima commented Mar 6, 2026 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Tracking issue

Why are the changes needed?

Alternatives

1. Store map of ActionEvents under TaskAction CR status

2. Store ActionEvent in a new CR

What changes were proposed in this pull request?

How was this patch tested?

Labels

Setup process

Screenshots

Check all the applicable boxes

Related PRs

Docs link

Uh oh!

Uh oh!

pingsutw Mar 9, 2026

Choose a reason for hiding this comment

Uh oh!

machichima Mar 9, 2026

Choose a reason for hiding this comment

Uh oh!

pingsutw Mar 9, 2026

Choose a reason for hiding this comment

Uh oh!

popojk left a comment

Choose a reason for hiding this comment

Uh oh!

popojk Mar 9, 2026

Choose a reason for hiding this comment

Uh oh!

machichima Mar 9, 2026

Choose a reason for hiding this comment

Uh oh!

popojk Mar 9, 2026

Choose a reason for hiding this comment

Uh oh!

machichima Mar 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

machichima Mar 9, 2026

Choose a reason for hiding this comment

Uh oh!

machichima commented Mar 10, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

machichima commented Mar 6, 2026 •

edited by github-actions bot

Loading

1. Store map of `ActionEvents` under `TaskAction` CR status

2. Store `ActionEvent` in a new CR

machichima Mar 9, 2026 •

edited

Loading