argoproj/argo-workflows · error

plugin %s open stream failed: %w

Error message

plugin %s open stream failed: %w

What it means

Driver.OpenStream calls the plugin's OpenStream RPC to get a server-stream of artifact bytes. This error wraps a failure establishing the stream (gRPC status error, connection dropped, context done) — no byte stream was created, so the caller receives nil reader.

Source

Thrown at workflow/artifacts/plugin/plugin.go:169

		Path:          path,
	})
	if err != nil {
		return fmt.Errorf("plugin %s load failed: %w", d.pluginName, err)
	}
	if !resp.Success {
		return fmt.Errorf("plugin %s load failed: %s", d.pluginName, resp.Error)
	}
	return nil
}

// OpenStream implements ArtifactDriver.OpenStream by calling the plugin service
func (d *Driver) OpenStream(ctx context.Context, a *wfv1.Artifact) (io.ReadCloser, error) {
	grpcArtifact := convertToGRPC(a)
	stream, err := d.client.OpenStream(ctx, &artifact.OpenStreamRequest{
		Artifact: grpcArtifact,
	})
	if err != nil {
		return nil, fmt.Errorf("plugin %s open stream failed: %w", d.pluginName, err)
	}

	reader, writer := io.Pipe()

	go func() {
		defer writer.Close()
		for {
			resp, err := stream.Recv()
			if errors.Is(err, io.EOF) {
				break
			}
			if err != nil {
				writer.CloseWithError(fmt.Errorf("plugin %s stream receive failed: %w", d.pluginName, err))
				return
			}
			if resp.Error != "" {
				writer.CloseWithError(fmt.Errorf("plugin %s stream error: %s", d.pluginName, resp.Error))
				return

View on GitHub (pinned to 35bff19146)

Solutions

1. Check plugin pod health and restart the workflow step if Unavailable
2. Inspect status.Code(err) on the wrapped error to distinguish transient vs permanent causes
3. Upgrade the plugin if it does not implement OpenStream (Unimplemented)
4. Raise the context timeout / connectionTimeoutSeconds for large streams

Example fix

// before
rc, err := driver.OpenStream(ctx, a)
// after
code := status.Code(err)
if code == codes.Unavailable || code == codes.DeadlineExceeded {
    rc, err = driver.OpenStream(ctx, a) // retry once with fresh ctx
}
if err != nil { return nil, fmt.Errorf("open stream: %w", err) }

// verify plugin readiness before opening streams
if err := drv.ping(ctx); err != nil { // e.g. GetCapabilities probe
    return fmt.Errorf("plugin not ready for streaming: %w", err)
}

rc, err := drv.OpenStream(ctx, a)
if err != nil {
    if c := status.Code(err); c == codes.Unavailable || c == codes.DeadlineExceeded {
        return retryWithBackoff(func() (io.ReadCloser, error) { return drv.OpenStream(ctx, a) })
    }
    if c == codes.Unimplemented {
        return fallbackToUnaryLoad(ctx, a) // old plugin without OpenStream
    }
    return nil, err
}

When it happens

Trigger: OpenStream RPC fails at call time: plugin Unavailable (restarting), DeadlineExceeded, Canceled context, Unimplemented (old plugin without OpenStream), or message-size/auth failures on the request.

Common situations: Streaming an artifact (e.g. to logs or another artifact) right when the plugin pod restarts; using an older plugin build lacking the streaming API; short ctx deadlines on large artifacts.

Related errors

• plugin %s load failed: %w
• plugin %s stream receive failed: %w
• failed to dial plugin %s at %q: %w
• plugin %s connection shutdown (socket=%q)
• timeout waiting for plugin %s connection to become ready, la

AI-assisted analysis of argoproj/argo-workflows@35bff19146 (2026-09-03). Data as JSON: /api/errors/9d9b0d184e8582e0. Report an issue: GitHub.