Publish raw events #151

tnsetting · 2021-02-02T13:58:13Z

TL;DR

Send raw flyte events to a pubsub topic. This is the first step for flyte lineage events. These events will be consumed and enriched to external by another thread or process. If the externalEvent part is not configured, a noop publisher will be created.

Example of config (the notifications bit is orthogonal):

notifications:
  type: gcp
  region: "my-region"
  gcp:
    projectId: "test"
  publisher:
    topicName: "topic"
  processor:
    queueName: "queue"
    accountId: "bar"
external_events.yaml |
  externalEvent:
    eventPublisher:
      topicName: "event_topic"
      eventTypes:
      - "all" // Or workflow, node, task. Or "*"

Type

Bug Fix
Feature
Plugin

Are all requirements met?

Complete description

How did you fix the bug, make the feature etc. Link to any design docs etc

Tracking Issue

https://github.com/lyft/flyte/issues/

Follow-up issue

NA
OR
https://github.com/lyft/flyte/issues/

tnsetting · 2021-02-04T20:30:50Z

@kumare3 or @katrogan Any comments for the PR? Would it make more sense to have separate config section for the event instead of inside the notification section? Thanks.

katrogan

looks good! just a few suggestions

katrogan · 2021-02-04T22:33:12Z

pkg/async/notifications/factory.go

@@ -148,3 +148,57 @@ func NewNotificationsPublisher(config runtimeInterfaces.NotificationsConfig, sco
 		return implementations.NewNoopPublish()
 	}
 }
+
+func NewEventsPublisher(config runtimeInterfaces.NotificationsConfig, scope promutils.Scope) interfaces.Publisher {
+	if config.ExternalEvent.EventPublisherConfig.TopicName == "" {


can we explicitly just use the Type of the Notificationsconfig. and if it's not no-op, error when Topic Name is empty?

katrogan · 2021-02-04T22:42:05Z

pkg/async/notifications/implementations/event_publisher.go

+	events        []string
+}
+
+func getSupportedEvents() map[string]string {


nit: instead of using a func you can actually initialize the map in golang as a var. see https://stackoverflow.com/questions/41078272/initializing-a-go-map-in-a-single-statement for an example

pkg/async/notifications/implementations/event_publisher.go

katrogan · 2021-02-04T22:46:07Z

pkg/async/notifications/implementations/event_publisher.go

+func NewEventsPublisher(pub pubsub.Publisher, scope promutils.Scope, eventTypes string) interfaces.Publisher {
+	supportedEvents := getSupportedEvents()
+
+	var eventList = make([]string, 0)


do you want to use a String set here instead since you only ever check for membership/

katrogan · 2021-02-04T22:52:17Z

pkg/async/notifications/implementations/event_publisher.go

+// The key is the notification type as defined as an enum.
+func (p *EventPublisher) Publish(ctx context.Context, notificationType string, msg proto.Message) error {
+	p.systemMetrics.PublishTotal.Inc()
+	logger.Debugf(ctx, "Publishing the following message [%s]", msg.String())


does String() return a human readable message? if not can we use ...[%+v]", msg) instead?

katrogan · 2021-02-04T22:53:07Z

pkg/async/notifications/implementations/event_publisher.go

+func newEventPublisherSystemMetrics(scope promutils.Scope) eventPublisherSystemMetrics {
+	return eventPublisherSystemMetrics{
+		Scope:        scope,
+		PublishTotal: scope.MustNewCounter("event_publish_total", "overall count of publish messages"),


i know it's all additive, but can we add a success counter too?

katrogan · 2021-02-04T22:56:54Z

pkg/runtime/interfaces/application_configuration.go

+	// The topic which events should be published, e.g. node, task, workflow
+	TopicName string `json:"topicName"`
+	// Event types: task, node, workflow executions
+	EventTypes string `json:"eventTypes"`


why not an array type?

Because I want to support the option to use all or * and using a single string makes it easier to filter on such cases.

but you can still do that with a single entry, no? I'd prefer to use conventional yaml list formatting vs our own string-split scheme

katrogan · 2021-02-04T22:59:15Z

Would it make more sense to have separate config section for the event instead of inside the notification section? Thanks.

Yes let's please do that! I think we can worry about refactoring the existing email notifications implementation separately from the egress event work you're doing. Eventually notifications can be handled only by a separately configured subscriber so it make sense to separate out the events config now!

katrogan · 2021-02-05T18:10:32Z

hey @tnsetting is this ready for re-review?

tnsetting · 2021-02-05T21:26:50Z

hey @tnsetting is this ready for re-review?

Yes, it is ready.

katrogan

sorry just a few more comments

katrogan · 2021-02-05T22:00:02Z

pkg/runtime/interfaces/application_configuration.go

+	// The topic which events should be published, e.g. node, task, workflow
+	TopicName string `json:"topicName"`
+	// Event types: task, node, workflow executions
+	EventTypes string `json:"eventTypes"`


but you can still do that with a single entry, no? I'd prefer to use conventional yaml list formatting vs our own string-split scheme

katrogan · 2021-02-05T22:04:13Z

pkg/async/notifications/implementations/event_publisher.go

+var workflowExecutionReq admin.WorkflowExecutionEventRequest
+
+var supportedEvents = map[string]string{
+	"task":     proto.MessageName(&taskExecutionReq),


sorry should have brought this up in the last round, but can we use shared constants here as map keys? perhaps https://github.com/flyteorg/flyteadmin/blob/master/pkg/common/entity.go could work

katrogan · 2021-02-05T22:06:15Z

pkg/async/notifications/implementations/event_publisher.go

+// The key is the notification type as defined as an enum.
+func (p *EventPublisher) Publish(ctx context.Context, notificationType string, msg proto.Message) error {
+	p.systemMetrics.PublishTotal.Inc()
+	logger.Debugf(ctx, "Publishing the following message [%+v]", msg)


nit: do you want to move this log statement below the check to shouldPublishEvent?

katrogan · 2021-02-05T22:10:47Z

pkg/async/notifications/implementations/event_publisher_test.go

+	testEventPublisher.FoundError = nil
+}
+
+func TestNewEventsPublisher_EventTypes(t *testing.T) {


Thank you for this very exhaustive test 👍

katrogan · 2021-02-05T22:12:36Z

pkg/manager/impl/node_execution_manager.go

@@ -4,6 +4,9 @@ import (
 	"context"
 	"strconv"

+	"github.com/gogo/protobuf/proto"


can we use github.com/golang/protobuf/proto instead? gogo behaves weirdly with oneofs

Sure. Never paid attention for that as intellij did that for me.

katrogan · 2021-02-08T17:05:14Z

pkg/async/notifications/implementations/event_publisher.go

+var workflowExecutionReq admin.WorkflowExecutionEventRequest
+
+const (
+	Task          = "task"


I'm still concerned this is a little brittle because here we define the strings statically but in the manager methods we use the proto message name as the event type. does that correspond to these raw strings like "task" or "node"?

katrogan

do you mind updating https://github.com/flyteorg/flyteadmin/blob/master/flyteadmin_config.yaml with an example of the config changes too?

katrogan

THANK YOU so much for this change

Signed-off-by: Haytham Abuelfutuh <[email protected]>

* Move scopes to config Signed-off-by: Haytham Abuelfutuh <[email protected]> * missed Signed-off-by: Haytham Abuelfutuh <[email protected]> * wip Signed-off-by: Haytham Abuelfutuh <[email protected]> * Update handlers.go Signed-off-by: Haytham Abuelfutuh <[email protected]> * go get propeller at v0.5.12 (#146) Signed-off-by: Haytham Abuelfutuh <[email protected]> * Update to using latest flytepropeller v0.5.13 (#148) Signed-off-by: Haytham Abuelfutuh <[email protected]> * Update propeller to 0.5.14 (#149) * Update propeller * cleanup Signed-off-by: Haytham Abuelfutuh <[email protected]> * Filter executions by user (#150) Signed-off-by: Haytham Abuelfutuh <[email protected]> * Update CI post migration Signed-off-by: Haytham Abuelfutuh <[email protected]> * Update codecov link Signed-off-by: Haytham Abuelfutuh <[email protected]> * Publish raw events (#151) Signed-off-by: Haytham Abuelfutuh <[email protected]> * fix test Signed-off-by: Haytham Abuelfutuh <[email protected]> * wip Signed-off-by: Haytham Abuelfutuh <[email protected]> * Fix token retrieval from cookies Signed-off-by: Haytham Abuelfutuh <[email protected]> * Fix unit tests and lint Signed-off-by: Haytham Abuelfutuh <[email protected]> * Move to const Signed-off-by: Haytham Abuelfutuh <[email protected]> * Revert Auth config Signed-off-by: Haytham Abuelfutuh <[email protected]> * Revert kube config path Signed-off-by: Haytham Abuelfutuh <[email protected]> * Use access token when posting to IdP Signed-off-by: Haytham Abuelfutuh <[email protected]> * Ignore refresh token error Signed-off-by: Haytham Abuelfutuh <[email protected]> * Fix tests Signed-off-by: Haytham Abuelfutuh <[email protected]> * Expose openId metadata endpoint and expose scopes in /config endpoint Signed-off-by: Haytham Abuelfutuh <[email protected]> * mod tidy Signed-off-by: Haytham Abuelfutuh <[email protected]> * rename Signed-off-by: Haytham Abuelfutuh <[email protected]> * lint Signed-off-by: Haytham Abuelfutuh <[email protected]> * unit tests Signed-off-by: Haytham Abuelfutuh <[email protected]> Co-authored-by: Yee Hing Tong <[email protected]> Co-authored-by: Katrina Rogan <[email protected]> Co-authored-by: tnsetting <[email protected]>

tnsetting added 2 commits January 27, 2021 14:16

Add publisher for events

9306023

sending events to pubsub

6fa0eab

tnsetting requested review from anandswaminathan, EngHabu, katrogan, kumare3 and wild-endeavor as code owners February 2, 2021 13:58

tnsetting changed the title ~~Publish task events~~ [WIP]Publish task events Feb 2, 2021

tnsetting added 2 commits February 2, 2021 15:27

Fix test and lint

3e44470

More fixes for testcases

33ed060

tnsetting changed the title ~~[WIP]Publish task events~~ Publish task events Feb 2, 2021

tnsetting changed the title ~~Publish task events~~ Publish raw events Feb 3, 2021

katrogan reviewed Feb 4, 2021

View reviewed changes

tnsetting added 2 commits February 5, 2021 16:18

Address comments and use new config section

7c2a395

Fix goimports

3b170a1

tnsetting force-pushed the publish_task_events branch from d10d5a0 to 3b170a1 Compare February 5, 2021 15:52

just for trigger the build

9224c6c

katrogan reviewed Feb 5, 2021

View reviewed changes

more comments fix

1f91d61

katrogan reviewed Feb 8, 2021

View reviewed changes

Add config example

e242acb

tnsetting requested a review from katrogan February 9, 2021 12:14

katrogan approved these changes Feb 9, 2021

View reviewed changes

katrogan merged commit 659635f into flyteorg:master Feb 9, 2021

EngHabu pushed a commit that referenced this pull request Feb 27, 2021

Publish raw events (#151)

327494d

Signed-off-by: Haytham Abuelfutuh <[email protected]>

tnsetting mentioned this pull request May 11, 2021

[Feature] Support publishing raw event than email body flyteorg/flyte#293

Closed

13 tasks

eapolinario pushed a commit that referenced this pull request Sep 6, 2023

Publish raw events (#151)

6205a2e

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Publish raw events #151

Publish raw events #151

tnsetting commented Feb 2, 2021 •

edited by wild-endeavor

Loading

tnsetting commented Feb 4, 2021 •

edited

Loading

katrogan left a comment

katrogan Feb 4, 2021

katrogan Feb 4, 2021

katrogan Feb 4, 2021

katrogan Feb 4, 2021

katrogan Feb 4, 2021

katrogan Feb 4, 2021

tnsetting Feb 5, 2021

katrogan Feb 5, 2021

katrogan commented Feb 4, 2021

katrogan commented Feb 5, 2021

tnsetting commented Feb 5, 2021

katrogan left a comment

katrogan Feb 5, 2021

katrogan Feb 5, 2021

katrogan Feb 5, 2021

katrogan Feb 5, 2021

katrogan Feb 5, 2021

tnsetting Feb 6, 2021

katrogan Feb 8, 2021

katrogan Feb 9, 2021

katrogan left a comment

katrogan left a comment

Publish raw events #151

Publish raw events #151

Conversation

tnsetting commented Feb 2, 2021 • edited by wild-endeavor Loading

TL;DR

Type

Are all requirements met?

Complete description

Tracking Issue

Follow-up issue

tnsetting commented Feb 4, 2021 • edited Loading

katrogan left a 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

katrogan commented Feb 4, 2021

katrogan commented Feb 5, 2021

tnsetting commented Feb 5, 2021

katrogan left a 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

katrogan left a comment

Choose a reason for hiding this comment

katrogan left a comment

Choose a reason for hiding this comment

tnsetting commented Feb 2, 2021 •

edited by wild-endeavor

Loading

tnsetting commented Feb 4, 2021 •

edited

Loading