Skip to main content

Event Proxying for External Triggers

When external components like the Flyte executor need to report task phase transitions or status changes, they do not communicate directly with the internal run management logic. Instead, Flyte uses the EventsProxyService as a gateway to bridge external triggers to internal service calls, ensuring that events are validated and correctly routed.

Architectural Role of the Events Proxy​

The EventsProxyService (defined in events/service/events_proxy_service.go) acts as a thin gRPC/Connect wrapper around an internal client. Its primary responsibility is to receive RecordRequest messages containing one or more ActionEvent objects and forward them to the InternalRunServiceClient.

This design decouples the event ingestion layer from the core run management logic. By providing a dedicated proxy service, Flyte can expose a stable external API for event reporting while allowing the internal RunService to reside on a different network or evolve independently.

The Record Workflow​

The core functionality of the proxy is implemented in the Record method. This method performs several critical checks before forwarding the data:

  1. Client Readiness: It ensures the runClient is initialized, returning connect.CodeFailedPrecondition if it is missing.
  2. Validation: It calls req.Msg.Validate() on the incoming message to ensure the payload adheres to the expected schema.
  3. Short-circuiting: If the request contains no events, the service returns a successful response immediately without making an internal network call.
  4. Synchronous Forwarding: It maps the RecordRequest to a RecordActionEventsRequest and calls the internal service.
func (s *EventsProxyService) Record(ctx context.Context, req *connect.Request[workflow.RecordRequest]) (*connect.Response[workflow.RecordResponse], error) {
	if s.runClient == nil {
		return nil, connect.NewError(connect.CodeFailedPrecondition, fmt.Errorf("run client is not initialized"))
	}
	if err := req.Msg.Validate(); err != nil {
		logger.Errorf(ctx, "invalid EventsProxyService.Record request: %v", err)
		return nil, connect.NewError(connect.CodeInvalidArgument, err)
	}
	if len(req.Msg.GetEvents()) == 0 {
		return connect.NewResponse(&workflow.RecordResponse{}), nil
	}

	recordEventReq := &workflow.RecordActionEventsRequest{Events: req.Msg.GetEvents()}
	recordActionResp, err := s.runClient.RecordActionEvents(ctx, connect.NewRequest(recordEventReq))
	if err != nil {
		logger.Warnf(ctx, "failed to forward action events to run service: %v", err)
		return nil, connect.NewError(connect.CodeInternal, err)
	}
    // ...
	return connect.NewResponse(&workflow.RecordResponse{}), nil
}

Service Initialization and Setup​

The service is initialized during the Flyte events service startup in events/setup.go. The setup process involves creating a workflowconnect.InternalRunServiceClient pointing to the RunServiceURL (defaulting to http://localhost:8090) and mounting the proxy handler onto the application's HTTP mux.

// From events/setup.go
runClient := workflowconnect.NewInternalRunServiceClient(http.DefaultClient, runServiceURL, connect.WithInterceptors(otelInterceptor))
eventsSvc := service.NewEventsProxyService(runClient)

path, handler := workflowconnect.NewEventsProxyServiceHandler(eventsSvc, connect.WithInterceptors(otelInterceptor))
sc.Mux.Handle(path, handler)

This setup ensures that the proxy is equipped with necessary OpenTelemetry interceptors for tracing and metrics before it begins handling traffic.

Event Reporting in the Executor​

A primary consumer of this service is the TaskActionReconciler located in executor/pkg/controller/taskaction_controller.go. When a task's phase changes (e.g., from Running to Succeeded), the reconciler builds an ActionEvent and uses the EventsProxyServiceClient to persist it.

// From executor/pkg/controller/taskaction_controller.go
actionEvent := r.buildActionEvent(ctx, newTaskAction, phaseInfo)
if _, err := r.eventsClient.Record(ctx, connect.NewRequest(&workflow.RecordRequest{
    Events: []*workflow.ActionEvent{actionEvent},
})); err != nil {
    r.Recorder.Eventf(
        newTaskAction,
        nil,
        corev1.EventTypeWarning,
        "ActionEventPublishFailed",
        "PublishingActionEvent",
        "Failed to persist action event %q: %v",
        actionEvent.GetId().GetName(),
        err,
    )
    // ...
}

If the proxy returns an error, the executor logs the failure and records a Kubernetes event, ensuring that failures in the event pipeline are visible to operators.

Implementation Tradeoffs​

The current implementation of EventsProxyService is strictly synchronous. While this ensures that the caller (like the executor) knows immediately if an event was successfully persisted, it also means that the latency of the internal RunService directly impacts the performance of the event-producing components. Additionally, the proxy does not implement internal retries; it relies on the underlying Connect client or the calling component to handle transient network failures.