feat(connector): record connector endpoint on LogContext

flyteidl2/core/execution.proto

@@ -113,6 +113,18 @@ message TaskLog {
 message LogContext {
   repeated PodLogContext pods = 1;
   string primary_pod_name = 2;
+
+  // Optional. Set by the connector plugin when an action is served by a connector. The dataplane
+  // dataproxy uses this endpoint to proxy log requests to the connector's GetTaskLogs RPC instead
+  // of streaming pod logs.
+  ConnectorLogContext connector = 3;
+}
+
+// ConnectorLogContext describes how to reach a connector that owns an action's logs.
+message ConnectorLogContext {
+  // gRPC endpoint of the connector deployment (e.g.
+  // "batch-job-connector.flytesnacks-development.svc.cluster.local:80").
+  string endpoint = 1;
 }
 
 // Contains metadata required to identify logs produces by a single pod

flyteplugins/go/tasks/plugins/webapi/connector/plugin.go

@@ -319,6 +319,20 @@ func (p *Plugin) Status(ctx context.Context, taskCtx webapi.StatusContext) (phas
 	}
 
 	taskInfo := &core.TaskInfo{Logs: logLinks, CustomInfo: resource.CustomInfo}
+
+	// Record the connector endpoint on the LogContext so the dataplane dataproxy can stream logs
+	// straight from the connector via AsyncConnectorService.GetTaskLogs (resource_meta is fetched
+	// separately from PluginStateService at log-stream time).
+	if metadata, ok := taskCtx.ResourceMeta().(ResourceMetaWrapper); ok {
+		if connector, err := p.getFinalConnector(metadata.TaskCategory, p.cfg, metadata.Domain); err == nil && connector != nil && connector.ConnectorDeployment != nil {
+			taskInfo.LogContext = &flyteIdl.LogContext{
+				Connector: &flyteIdl.ConnectorLogContext{
+					Endpoint: connector.ConnectorDeployment.Endpoint,
+				},
+			}
+		}
+	}
+
 	errorCode := pluginErrors.TaskFailedWithError
 
 	switch resource.Phase {