argoproj/argo-workflows · error

plugin %s stream error: %s

Error message

plugin %s stream error: %s

What it means

This error surfaces when the artifact plugin itself reports an error message inside an otherwise-healthy OpenStream gRPC response: the driver is receiving chunks from the plugin's stream and a response carries a non-empty resp.Error field. The driver wraps the plugin's own error string and injects it into the io.Pipe via CloseWithError, so the caller of Load/any reader of the returned ReadCloser sees it as a read error. It means the plugin started streaming but failed on its side (e.g. could not read the source storage).

Source

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

	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
			}
			if resp.IsEnd {
				break
			}
			if len(resp.Data) > 0 {
				if _, writeErr := writer.Write(resp.Data); writeErr != nil {
					writer.CloseWithError(fmt.Errorf("plugin %s stream write failed: %w", d.pluginName, writeErr))
					return
				}
			}
		}
	}()

	return reader, nil
}

// Save implements ArtifactDriver.Save by calling the plugin service

View on GitHub (pinned to 35bff19146)

Solutions

  1. Read the embedded plugin error message after 'stream error: ' — it is the plugin's own diagnosis and names the real cause
  2. Check the plugin's logs (kubectl logs on the plugin container) for the corresponding failure
  3. Verify the artifact's source location/credentials are valid and accessible from the plugin pod
  4. Retry the artifact load once storage issues are fixed
Defensive patterns

Strategy: try-catch

Validate before calling

// No pre-call validation can predict plugin-internal failures; ensure the plugin is reachable first:
if _, err := os.Stat(socketPath); err != nil {
    return fmt.Errorf("plugin socket %s missing before OpenStream: %w", socketPath, err)
}

Type guard

func pluginStreamError(err error) (string, bool) {
    var pe *io.PipeError // or match on message prefix
    if errors.As(err, &pe) && strings.Contains(err.Error(), "stream error: ") {
        return err.Error(), true
    }
    return "", false
}

Try / catch

reader, err := driver.OpenStream(ctx, artifact)
if err != nil { return err }
if _, copyErr := io.Copy(dst, reader); copyErr != nil {
    if strings.Contains(copyErr.Error(), "stream error: ") {
        // plugin-reported failure: inspect plugin logs, fix storage creds/config, retry
    }
    reader.Close()
    return copyErr
}
reader.Close()

Prevention

When it happens

Trigger: Calling Driver.OpenStream (artifact plugin driver, workflow/artifacts/plugin/plugin.go:163) where the plugin's stream.Recv returns a response with resp.Error set — i.e. the plugin's gRPC server hit an error mid-stream and reported it in-band rather than as a gRPC status.

Common situations: Plugin cannot access the backing storage (bad S3/GCS credentials configured on the plugin), artifact key does not exist, plugin-side timeout while reading a large artifact, or plugin misconfigured storage backend.

Related errors

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