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

flyteidl2/dataproxy/dataproxy_service.proto

@@ -205,6 +205,11 @@ message TailLogsRequest {
 
   // The pod name to tail logs from. If not provided, attempt to find the primary pod, else assume the first pod.
   string pod_name = 3;
+
+  // +optional. Connector endpoint copied from ActionAttempt.log_context.connector.endpoint when the
+  // action is served by a connector. The dataproxy uses this to stream logs from the connector via
+  // AsyncConnectorService.GetTaskLogs instead of pod logs.
+  string connector_endpoint = 4;
 }
 
 // Reponse message for tailing logs.

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

@@ -13,17 +13,16 @@ import (
 	"google.golang.org/protobuf/types/known/structpb"
 	"k8s.io/apimachinery/pkg/util/wait"
 
-	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/secret"
-	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/utils"
-
 	pluginErrors "github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/errors"
 	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/logs"
 	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery"
 	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/core"
 	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/core/template"
 	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/io"
 	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/ioutils"
+	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/secret"
 	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/tasklog"
+	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/utils"
 	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/webapi"
 	"github.com/flyteorg/flyte/v2/flytestdlib/logger"
 	"github.com/flyteorg/flyte/v2/flytestdlib/promutils"
@@ -80,13 +79,14 @@ type Plugin struct {
 }
 
 type ResourceWrapper struct {
-	Phase          flyteIdl.TaskExecution_Phase
-	Outputs        *task.Outputs
-	Message        string
-	LogLinks       []*flyteIdl.TaskLog
-	CustomInfo     *structpb.Struct
-	ConnectorID    string
-	IsConnectorApp bool
+	Phase             flyteIdl.TaskExecution_Phase
+	Outputs           *task.Outputs
+	Message           string
+	LogLinks          []*flyteIdl.TaskLog
+	CustomInfo        *structpb.Struct
+	ConnectorID       string
+	IsConnectorApp    bool
+	ConnectorEndpoint string
 }
 
 // IsTerminal is used to avoid making network calls to the connector service if the resource is already in a terminal state.
@@ -244,15 +244,19 @@ func (p *Plugin) Get(ctx context.Context, taskCtx webapi.GetContext) (latest web
 		return nil, fmt.Errorf("failed to get task from connector with %v", err)
 	}
 
-	return ResourceWrapper{
+	wrapper := ResourceWrapper{
 		Phase:          res.GetResource().GetPhase(),
 		Outputs:        res.GetResource().GetOutputs(),
 		Message:        res.GetResource().GetMessage(),
 		LogLinks:       res.GetResource().GetLogLinks(),
 		CustomInfo:     res.GetResource().GetCustomInfo(),
 		ConnectorID:    connector.ConnectorID,
 		IsConnectorApp: connector.IsConnectorApp,
-	}, nil
+	}
+	if connector.ConnectorDeployment != nil {
+		wrapper.ConnectorEndpoint = connector.ConnectorDeployment.Endpoint
+	}
+	return wrapper, nil
 }
 
 func (p *Plugin) Delete(ctx context.Context, taskCtx webapi.DeleteContext) error {
@@ -319,6 +323,18 @@ func (p *Plugin) Status(ctx context.Context, taskCtx webapi.StatusContext) (phas
 	}
 
 	taskInfo := &core.TaskInfo{Logs: logLinks, CustomInfo: resource.CustomInfo}
+
+	if resource.ConnectorEndpoint != "" {
+		taskInfo.LogContext = &flyteIdl.LogContext{
+			Connector: &flyteIdl.ConnectorLogContext{
+				Endpoint: resource.ConnectorEndpoint,
+			},
+		}
+		logger.Infof(ctx, "Recorded connector endpoint %q on LogContext for connector %q", resource.ConnectorEndpoint, resource.ConnectorID)
+	} else {
+		logger.Debugf(ctx, "Connector endpoint missing on resource for connector %q; LogContext.connector not set", resource.ConnectorID)
+	}
+
 	errorCode := pluginErrors.TaskFailedWithError
 
 	switch resource.Phase {