wip

Signed-off-by: Eduardo Apolinario <[email protected]>

flyteplugins/go/tasks/pluginmachinery/core/exec_context.go

@@ -63,6 +63,9 @@ type TaskExecutionContext interface {
 
 	// Returns a handle to the Task events recorder, which get stored in the Admin.
 	EventsRecorder() EventsRecorder
+
+	// TODO: Is this right? Where else can I plop this in?
+	// ConsoleURL() string
 }
 
 // A simple fire-and-forget func

flyteplugins/go/tasks/pluginmachinery/core/exec_metadata.go

@@ -32,6 +32,8 @@ type TaskExecutionID interface {
 	// GetUniqueNodeID returns the fully-qualified Node ID that is unique within a
 	// given workflow execution.
 	GetUniqueNodeID() string
+
+	GetConsoleURL() string
 }
 
 // TaskExecutionMetadata represents any execution information for a Task. It is used to communicate meta information about the

flyteplugins/go/tasks/pluginmachinery/flytek8s/k8s_resource_adds.go

@@ -2,6 +2,7 @@ package flytek8s
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"strconv"
 
@@ -69,6 +70,15 @@ func GetExecutionEnvVars(id pluginsCore.TaskExecutionID) []v1.EnvVar {
 		// },
 	}
 
+	if id.GetConsoleURL() != "" {
+		envVars = append(envVars, v1.EnvVar{
+			Name: "FLYTE_EXECUTION_URL",
+			// TODO: is it safe to append `attempt` like this?
+			// TODO: should we use net/url to build this url?
+			Value: fmt.Sprintf("%s/projects/%s/domains/%s/executions/%s/nodeId/%s-%s/nodes", id.GetConsoleURL(), nodeExecutionID.Project, nodeExecutionID.Domain, nodeExecutionID.Name, id.GetUniqueNodeID(), attemptNumber),
+		})
+	}
+
 	// Task definition Level env variables.
 	if id.GetID().TaskId != nil {
 		taskID := id.GetID().TaskId

flytepropeller/cmd/kubectl-flyte/cmd/printers/workflow.go

@@ -47,6 +47,8 @@ func (w *ContextualWorkflow) GetExecutionConfig() v1alpha1.ExecutionConfig {
 	return v1alpha1.ExecutionConfig{}
 }
 
+func (w *ContextualWorkflow) GetConsoleURL() string { return "" }
+
 type WorkflowPrinter struct {
 }
 

flytepropeller/pkg/apis/flyteworkflow/v1alpha1/iface.go

@@ -511,6 +511,7 @@ type Meta interface {
 	GetEventVersion() EventVersion
 	GetDefinitionVersion() WorkflowDefinitionVersion
 	GetRawOutputDataConfig() RawOutputDataConfig
+	GetConsoleURL() string
 }
 
 type TaskDetailsGetter interface {
@@ -534,6 +535,7 @@ type ExecutableWorkflow interface {
 	MetaExtended
 	NodeStatusGetter
 	GetExecutionConfig() ExecutionConfig
+	GetConsoleURL() string
 }
 
 type NodeStatusGetter interface {

flytepropeller/pkg/controller/executors/execution_context.go

@@ -19,6 +19,7 @@ type ImmutableExecutionContext interface {
 	GetID() v1alpha1.WorkflowID
 	GetOnFailurePolicy() v1alpha1.WorkflowOnFailurePolicy
 	GetExecutionConfig() v1alpha1.ExecutionConfig
+	GetConsoleURL() string
 }
 
 type ParentInfoGetter interface {

flytepropeller/pkg/controller/nodes/executor.go

@@ -500,6 +500,7 @@ type nodeExecutor struct {
 	shardSelector                   ioutils.ShardSelector
 	store                           *storage.DataStore
 	taskRecorder                    events.TaskEventRecorder
+	consoleURL                      string
 }
 
 func (c *nodeExecutor) RecordTransitionLatency(ctx context.Context, dag executors.DAGStructure, nl executors.NodeLookup, node v1alpha1.ExecutableNode, nodeStatus v1alpha1.ExecutableNodeStatus) {

flytepropeller/pkg/controller/nodes/interfaces/node_exec_context.go

@@ -39,6 +39,7 @@ type NodeExecutionMetadata interface {
 	GetSecurityContext() core.SecurityContext
 	IsInterruptible() bool
 	GetInterruptibleFailureThreshold() int32
+	GetConsoleURL() string
 }
 
 type NodeExecutionContext interface {

flytepropeller/pkg/controller/nodes/node_exec_context.go

@@ -118,6 +118,8 @@ func (e nodeExecMetadata) GetLabels() map[string]string {
 	return e.nodeLabels
 }
 
+func (e nodeExecMetadata) GetConsoleURL() string { return e.Meta.GetConsoleURL() }
+
 type nodeExecContext struct {
 	store           *storage.DataStore
 	tr              interfaces.TaskReader

flytepropeller/pkg/controller/nodes/task/taskexec_context.go

@@ -36,6 +36,7 @@ type taskExecutionID struct {
 	execName     string
 	id           *core.TaskExecutionIdentifier
 	uniqueNodeID string
+	consoleURL   string
 }
 
 func (te taskExecutionID) GetID() core.TaskExecutionIdentifier {
@@ -63,6 +64,8 @@ func (te taskExecutionID) GetGeneratedNameWith(minLength, maxLength int) (string
 	return te.execName, nil
 }
 
+func (te taskExecutionID) GetConsoleURL() string { return te.consoleURL }
+
 type taskExecutionMetadata struct {
 	interfaces.NodeExecutionMetadata
 	taskExecID           taskExecutionID
@@ -290,6 +293,8 @@ func (t *Handler) newTaskExecutionContext(ctx context.Context, nCtx interfaces.N
 		return nil, err
 	}
 
+	nCtx.Node().GetTaskID()
+
 	return &taskExecutionContext{
 		NodeExecutionContext: nCtx,
 		tm: taskExecutionMetadata{
@@ -298,6 +303,7 @@ func (t *Handler) newTaskExecutionContext(ctx context.Context, nCtx interfaces.N
 				execName:     uniqueID,
 				id:           id,
 				uniqueNodeID: currentNodeUniqueID,
+				consoleURL:   nCtx.NodeExecutionMetadata().GetConsoleURL(),
 			},
 			o:                    nCtx.Node(),
 			maxAttempts:          maxAttempts,