Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 28 additions & 4 deletions cmd/wfctl/pipeline_docker_unix.go
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,13 @@ type pipelineDockerProcessGroupKey struct{}

var pipelineDockerTLSNames = [3]string{"ca.pem", "cert.pem", "key.pem"}

func resolvePipelineDockerClient(ctx context.Context) (*pipelineDockerClient, error) {
func resolvePipelineDockerClient(ctx context.Context) (_ *pipelineDockerClient, resolveErr error) {
phase := "docker_resolve_cli"
defer func() {
if resolveErr != nil {
pipelineRecordDiagnosticFromContext(ctx).fail(phase)
}
}()
// Only discovery receives a snapshot of Docker's transport-selection env.
// All daemon calls use a private empty config, explicit host/TLS flags, and
// a minimal env, avoiding auth helpers, config proxies, and context changes.
Expand All @@ -78,10 +84,12 @@ func resolvePipelineDockerClient(ctx context.Context) (*pipelineDockerClient, er
if err != nil {
return nil, fmt.Errorf("pipeline Docker CLI unavailable")
}
phase = "docker_resolve_home"
home := snapshot["HOME"]
if home == "" {
return nil, fmt.Errorf("pipeline Docker home unavailable")
}
phase = "docker_resolve_config"
config := snapshot["DOCKER_CONFIG"]
if config == "" {
config = filepath.Join(home, ".docker")
Expand All @@ -97,14 +105,17 @@ func resolvePipelineDockerClient(ctx context.Context) (*pipelineDockerClient, er
discoveryEnv = append(discoveryEnv, key+"="+value)
}
}
out, err := runPipelineDockerCLI(ctx, binary, discoveryEnv, pipelineDockerCallTimeout, "--config", config, "context", "inspect")
phase = "docker_resolve_context"
out, err := runPipelineDockerCLI(pipelineDockerResolutionContext(ctx, "resolve_context"), binary, discoveryEnv, pipelineDockerCallTimeout, "--config", config, "context", "inspect")
if err != nil {
return nil, fmt.Errorf("pipeline Docker context resolution failed")
}
phase = "docker_resolve_context_metadata"
var contexts []pipelineDockerContext
if json.Unmarshal([]byte(out.Stdout), &contexts) != nil || len(contexts) != 1 {
return nil, fmt.Errorf("pipeline Docker context metadata invalid")
}
phase = "docker_resolve_endpoint"
metadata := contexts[0]
endpoint, ok := metadata.Endpoints["docker"]
if !ok || metadata.Name == "" {
Expand All @@ -113,6 +124,7 @@ func resolvePipelineDockerClient(ctx context.Context) (*pipelineDockerClient, er
if err := validatePipelineDockerEndpoint(endpoint.Host); err != nil {
return nil, err
}
phase = "docker_resolve_tls"
var material [3][]byte
tlsEnabled := false
verifyTLS := false
Expand Down Expand Up @@ -153,6 +165,7 @@ func resolvePipelineDockerClient(ctx context.Context) (*pipelineDockerClient, er
if (len(material[1]) == 0) != (len(material[2]) == 0) {
return nil, fmt.Errorf("pipeline Docker TLS client material incomplete")
}
phase = "docker_resolve_private_config"
dir, err := os.MkdirTemp("", "wfctl-pipeline-docker-")
if err != nil {
return nil, fmt.Errorf("pipeline Docker private config unavailable")
Expand Down Expand Up @@ -182,7 +195,8 @@ func resolvePipelineDockerClient(ctx context.Context) (*pipelineDockerClient, er
c.args = append(c.args, flag+"="+path)
}
}
serverID, err := c.serverID(ctx)
phase = "docker_resolve_identity_metadata"
serverID, err := c.serverID(pipelineDockerResolutionContext(ctx, "resolve_identity"))
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -426,13 +440,23 @@ func runPipelineDockerCLI(ctx context.Context, binary string, env []string, time
if cmd.ProcessState != nil {
result.ExitCode = cmd.ProcessState.ExitCode()
}
if ctx.Err() != nil {
if contextErr := ctx.Err(); contextErr != nil {
outcome := "command"
switch contextErr {
case context.DeadlineExceeded:
outcome = "deadline"
case context.Canceled:
outcome = "cancelled"
}
observePipelineDockerResolutionFailure(ctx, outcome, result.ExitCode)
return result, fmt.Errorf("pipeline Docker operation canceled or timed out")
}
if stdout.overflow || stderr.overflow {
observePipelineDockerResolutionFailure(ctx, "capture", result.ExitCode)
return result, fmt.Errorf("pipeline Docker output exceeded capture limit")
}
if err != nil {
observePipelineDockerResolutionFailure(ctx, "command", result.ExitCode)
return result, fmt.Errorf("pipeline Docker operation failed")
}
return result, nil
Expand Down
56 changes: 56 additions & 0 deletions cmd/wfctl/pipeline_record_diagnostic_unix.go
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,14 @@ var pipelineRecordDiagnosticPhases = map[string]bool{
"container_create": true, "container_id": true, "container_attach": true,
"container_wait": true, "container_status": true, "container_command": true,
"cleanup": true, "pipeline_result": true, "record_write": true,
"docker_resolve_cli": true, "docker_resolve_home": true, "docker_resolve_config": true,
"docker_resolve_context": true, "docker_resolve_context_metadata": true,
"docker_resolve_endpoint": true, "docker_resolve_tls": true,
"docker_resolve_private_config": true, "docker_resolve_identity_metadata": true,
"docker_resolve_context_cancelled": true, "docker_resolve_context_deadline": true,
"docker_resolve_context_capture": true, "docker_resolve_context_command": true,
"docker_resolve_identity_cancelled": true, "docker_resolve_identity_deadline": true,
"docker_resolve_identity_capture": true, "docker_resolve_identity_command": true,
}

type pipelineRecordDiagnostic struct {
Expand All @@ -63,6 +71,54 @@ func pipelineRecordDiagnosticFromContext(ctx context.Context) *pipelineRecordDia
return diagnostic
}

// The private tag applies only to the two initial resolver calls. Subsequent
// server-ID checks and successful resolution never set operation/status fields.
type pipelineDockerResolutionKey struct{}

func pipelineDockerResolutionContext(ctx context.Context, operation string) context.Context {
if pipelineRecordDiagnosticFromContext(ctx) == nil {
return ctx
}
switch operation {
case "resolve_context", "resolve_identity":
return context.WithValue(ctx, pipelineDockerResolutionKey{}, operation)
default:
return ctx
}
}

func observePipelineDockerResolutionFailure(ctx context.Context, outcome string, exit int) {
diagnostic := pipelineRecordDiagnosticFromContext(ctx)
if diagnostic == nil {
return
}
operation, _ := ctx.Value(pipelineDockerResolutionKey{}).(string)
var stage string
switch operation {
case "resolve_context":
stage = "context"
case "resolve_identity":
stage = "identity"
default:
return
}
switch outcome {
case "cancelled", "deadline", "capture", "command":
default:
return
}
diagnostic.mu.Lock()
defer diagnostic.mu.Unlock()
if diagnostic.report.Phase == "" {
// Freeze the whole observation together; later cleanup or operations
// cannot mix their status with the first resolver failure.
diagnostic.report.Phase = "docker_resolve_" + stage + "_" + outcome
diagnostic.report.Operation = operation
diagnostic.report.DockerExit = pipelineRecordDiagnosticExit(exit)
diagnostic.report.ContainerExit = -1
}
}

func (d *pipelineRecordDiagnostic) fail(phase string) {
if d == nil {
return
Expand Down
Loading
Loading