logstream

package
v3.7.1 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Sep 8, 2026 License: Apache-2.0 Imports: 21 Imported by: 0

Documentation

Index

Constants

View Source
const SseTerminalDrainDelay = 5 * time.Second

SseTerminalDrainDelay is how long we wait after the pipeline reaches a terminal state before canceling SSE streams, to allow straggling log events to arrive.

Variables

View Source
var BaseSkipStepTypes = map[string]bool{
	"PIPELINE_SECTION":        true,
	"STAGES_STEP":             true,
	"NG_EXECUTION":            true,
	"STRATEGY":                true,
	"IntegrationStageStepPMS": true,
}

BaseSkipStepTypes are container nodes with no meaningful log content. Callers can extend with their own module-specific types via FollowMulti's extraSkipTypes param.

Functions

func ExecIdFromArg

func ExecIdFromArg(args []string) string

ExecIdFromArg extracts the bare execId from args[0] which may be "pipelineId/execId" or just "execId".

func FetchAndPrintLog

func FetchAndPrintLog(hc *http.Client, a *auth.ResolvedAuth, logKey, fmtFlag string, isPty bool, out io.Writer) (bool, error)

FetchAndPrintLog fetches a log blob and writes it to out. Returns (false, nil) when the blob exists but is empty.

func FetchAndPrintLogWithRetry

func FetchAndPrintLogWithRetry(hc *http.Client, a *auth.ResolvedAuth, shortKey, fmtFlag string, isPty bool, out io.Writer, endTs int64) (bool, error)

FetchAndPrintLogWithRetry wraps FetchAndPrintLog with retries for recently-completed steps whose blob may not yet be written. endTs is the step's end timestamp in milliseconds; if the blob is empty and the step finished within 60s, it retries up to 3 times (2s apart).

func FetchBlobToChannel

func FetchBlobToChannel(hc *http.Client, a *auth.ResolvedAuth, shortKey, source, fmtFlag string, isPty bool, ch chan<- Event, endTs int64) (bool, error)

FetchBlobToChannel fetches a completed log blob and sends it as a single EvBlob event. endTs is the step's end timestamp in milliseconds, used to decide whether to retry on empty.

func FindNodeForFollow

func FindNodeForFollow(g execgraph.ExecutionGraph, uuid string) (execgraph.GraphNode, bool)

FindNodeForFollow finds the graph node with the given UUID.

func FmtDuration

func FmtDuration(ms int64) string

FmtDuration formats a millisecond duration for event lines.

func FollowMulti

func FollowMulti(ctx *cmdctx.Ctx, execId, stageFilter, stepFilter string, style MultiStyle, extraSkipTypes map[string]bool) error

FollowMulti follows all log keys for an execution (or a stage/step filtered subset). extraSkipTypes supplements BaseSkipStepTypes — pass nil to use the base set only.

func IsTerminalStatus

func IsTerminalStatus(status string) bool

IsTerminalStatus returns true when a step or pipeline has reached a final state.

func RenderLogLinesToWriter

func RenderLogLinesToWriter(text, fmtFlag string, isPty bool, w io.Writer) error

func ShortLogKey

func ShortLogKey(key string) string

ShortLogKey strips the leading accountId/pipeline/ prefix from a logBaseKey.

func StreamSSEToChannel

func StreamSSEToChannel(ctx context.Context, hc *http.Client, a *auth.ResolvedAuth, logKey, source, fmtFlag string, isPty bool, ch chan<- Event) (bool, error)

StreamSSEToChannel opens the log-service SSE stream and sends each rendered log frame as an EvLogLine event. Returns (hadContent, error): hadContent is true if at least one log line was sent to ch. Returns when the server closes with an eof event, on error, or when ctx is cancelled.

func WriteEndEvent

func WriteEndEvent(w io.Writer, source string, node execgraph.GraphNode)

WriteEndEvent writes the <<< end event line for a node.

func WriteEvents

func WriteEvents(ch <-chan Event, w io.Writer, style MultiStyle)

WriteEvents drains ch and writes events to w.

Types

type Event

type Event struct {
	Kind    EventKind
	Source  string
	StartTs int64               // EvStart
	Node    execgraph.GraphNode // EvEnd
	Lines   []string            // EvLogLine / EvBlob
}

type EventKind

type EventKind int
const (
	EvStart   EventKind = iota // >>> started
	EvEnd                      // <<< completed / failed / aborted / expired
	EvLogLine                  // single rendered log line (from SSE, one per frame)
	EvBlob                     // full blob content sent as one chunk
)

type LogKeyEntry

type LogKeyEntry struct {
	LogKey     string
	Name       string
	FQN        string
	Status     string
	Depth      int
	ParentName string
	StartTs    int64
	EndTs      int64
	Delegates  []string // delegate names from delegateInfoList
	Inputs     string   // raw JSON from stepParameters
	Outputs    string   // raw JSON from outcomes
}

func FetchLogKeys

func FetchLogKeys(ctx *cmdctx.Ctx, execId string) ([]LogKeyEntry, string, error)

FetchLogKeys fetches the execution graph for execId and returns all short log keys along with the pipeline's current status string.

func StageSubtreeEntries

func StageSubtreeEntries(entries []LogKeyEntry, stageFilter string) []LogKeyEntry

StageSubtreeEntries filters entries down to the full subtree of steps nested under the stage named stageFilter, excluding the stage's own top-level entry. Subtree membership is determined via BaseFQN (a dot-delimited ancestry path), not immediate-parent name, so steps nested behind containers like spec.execution or a parallel step-group are included. Returns nil if no entry's name matches stageFilter. stageFilter == "" returns entries unchanged.

type MultiStyle

type MultiStyle int
const (
	MultiStyleMarkers MultiStyle = iota // ==> source switching markers + events (default)
	MultiStyleInline                    // [source] prefix on each log line, no markers/events
)

func ParseMultiStyle

func ParseMultiStyle(s string) MultiStyle

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL