feat: service logs command to stream service logs (#196)

* chore: fix landing logo shadow

* feat: implement service logs command with colored output and strict ordering

* revert Makefile

* fix after rebase

* move logs command under services with root shortcut

* simplify server options for streaming logs and CLI flags

* refactor ContainerLogs grpc server

* update --tail flag

* minor docker.proto

* refactor client ServiceLogs and ContainerLogs

* move ProxyMachinesContext from api to client pkg

* minor refactor proto Stream

* add api/logs

* implement LogMerger

* fix LogMerger to correctly use semaphore

* increate inflish entries to 100 per stream

* send heartbeats

* refactor ContainerLogs to synchronise Send of entries and heartbeats to the steram

* minor logmerege

* remove ContainerName form ServiceLogEntryMetadata

* refactor ContainerLogs into docker.Service

* detect stalled container logs streams

* update logmerger tests

* fix comment in test

* refactor LogMerger with options

* uc logs: format one or multiple services

* make LogMerger emit heartbeats, emit entries <= watermark, rewrite tests

* update uc logs with new LogMerger

* go mod tidy

* fix after merge

---------

Co-authored-by: Evgenii Orlov <evgenii.orlov@semrush.com>

pkg/client/client.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"strings"
 
 	"github.com/docker/cli/cli/streams"
 	"github.com/psviderski/uncloud/internal/machine/api/pb"
@@ -76,3 +77,41 @@ func proxyToMachine(ctx context.Context, machine *pb.MachineInfo) context.Contex
 	md := metadata.Pairs("machines", machineIP.String())
 	return metadata.NewOutgoingContext(ctx, md)
 }
+
+// ProxyMachinesContext returns a new context that proxies gRPC requests to the specified machines.
+// If namesOrIDs is nil, all machines are included.
+func (cli *Client) ProxyMachinesContext(
+	ctx context.Context, namesOrIDs []string,
+) (context.Context, api.MachineMembersList, error) {
+	// TODO: move the machine IP resolution to the proxy router to allow setting machine names and IDs in the metadata.
+	machines, err := cli.ListMachines(ctx, nil)
+	if err != nil {
+		return nil, nil, fmt.Errorf("list machines: %w", err)
+	}
+
+	var proxiedMachines api.MachineMembersList
+	var notFound []string
+	for _, nameOrID := range namesOrIDs {
+		if m := machines.FindByNameOrID(nameOrID); m != nil {
+			proxiedMachines = append(proxiedMachines, m)
+		} else {
+			notFound = append(notFound, nameOrID)
+		}
+	}
+
+	if len(notFound) > 0 {
+		return nil, nil, fmt.Errorf("machines not found: %s", strings.Join(notFound, ", "))
+	}
+
+	if len(namesOrIDs) == 0 {
+		proxiedMachines = machines
+	}
+
+	md := metadata.New(nil)
+	for _, m := range proxiedMachines {
+		machineIP, _ := m.Machine.Network.ManagementIp.ToAddr()
+		md.Append("machines", machineIP.String())
+	}
+
+	return metadata.NewOutgoingContext(ctx, md), proxiedMachines, nil
+}

pkg/client/image.go

@@ -48,7 +48,7 @@ func (cli *Client) InspectRemoteImage(ctx context.Context, id string) ([]api.Mac
 // it lists images on all machines.
 func (cli *Client) ListImages(ctx context.Context, filter api.ImageFilter) ([]api.MachineImages, error) {
 	// Broadcast the image list request to the specified machines or all machines if none specified.
-	listCtx, machines, err := api.ProxyMachinesContext(ctx, cli, filter.Machines)
+	listCtx, machines, err := cli.ProxyMachinesContext(ctx, filter.Machines)
 	if err != nil {
 		return nil, fmt.Errorf("create request context to broadcast to machines: %w", err)
 	}

pkg/client/logmerger.go

@@ -0,0 +1,310 @@
+package client
+
+import (
+	"container/heap"
+	"sync"
+	"time"
+
+	"github.com/psviderski/uncloud/pkg/api"
+)
+
+// logMergerMaxInFlightPerStream limits how many entries each input stream can have in the processing queue before being
+// throttled. This ensures fair interleaving between streams and prevents one fast stream from causing unbounded
+// buffering while waiting for slower streams.
+const (
+	logMergerMaxInFlightPerStream = 100
+	// logMergerHeartbeatDebounceInterval defines the minimum interval between deduplicated heartbeat entries emitted
+	// by the LogMerger. Keep it in sync with logsHeartbeatInterval internal/machine/docker/server.go.
+	logMergerHeartbeatDebounceInterval = 200 * time.Millisecond
+)
+
+// LogMergerOptions configures the behavior of LogMerger.
+type LogMergerOptions struct {
+	// StallTimeout specifies how long a stream can go without receiving any data before it's considered
+	// stalled and excluded from watermark calculation. A zero timeout disables stall detection.
+	StallTimeout time.Duration
+	// StallCheckInterval specifies how often to check for stalled streams.
+	StallCheckInterval time.Duration
+}
+
+// DefaultLogMergerOptions provides sensible default options that enable stall detection for LogMerger.
+var DefaultLogMergerOptions = LogMergerOptions{
+	StallTimeout:       10 * time.Second,
+	StallCheckInterval: 1 * time.Second,
+}
+
+// LogMerger merges multiple log streams into a single chronologically ordered stream based on timestamps.
+// It uses a low watermark algorithm to ensure proper ordering across streams.
+// Heartbeat entries from streams advance the watermark to enable timely emission of buffered logs.
+type LogMerger struct {
+	streams []*mergerStream
+	queue   logsHeap
+	// watermark is min(latest_timestamp for each stream).
+	watermark time.Time
+	output    chan api.ServiceLogEntry
+	// lastEmitted is the timestamp of the last emitted log entry or heartbeat.
+	lastEmitted        time.Time
+	stallTimeout       time.Duration
+	stallCheckInterval time.Duration
+}
+
+// NewLogMerger creates a new LogMerger for the given input streams with the specified options.
+func NewLogMerger(streams []<-chan api.ServiceLogEntry, opts LogMergerOptions) *LogMerger {
+	mergerStreams := make([]*mergerStream, len(streams))
+	now := time.Now()
+	for i, ch := range streams {
+		mergerStreams[i] = &mergerStream{
+			stream:       ch,
+			semaphore:    make(chan struct{}, logMergerMaxInFlightPerStream),
+			lastActivity: now,
+		}
+	}
+
+	return &LogMerger{
+		streams:            mergerStreams,
+		output:             make(chan api.ServiceLogEntry),
+		stallTimeout:       opts.StallTimeout,
+		stallCheckInterval: opts.StallCheckInterval,
+	}
+}
+
+// Stream starts the merge process and returns a channel that emits log entries in chronological order.
+// The returned channel is closed when all input streams are closed.
+func (m *LogMerger) Stream() <-chan api.ServiceLogEntry {
+	if len(m.streams) == 0 {
+		close(m.output)
+		return m.output
+	}
+
+	go m.run()
+
+	return m.output
+}
+
+// mergerStream combines a stream channel with its state and flow control.
+type mergerStream struct {
+	stream    <-chan api.ServiceLogEntry
+	semaphore chan struct{}
+	// Latest timestamp seen from this stream (log or heartbeat).
+	lastSeen time.Time
+	// Wall clock time when we last received any data from this stream.
+	lastActivity time.Time
+	// Metadata associated with this stream. It's populated from the first log entry received.
+	metadata *api.ServiceLogEntryMetadata
+	// Whether the stream channel has closed.
+	closed bool
+	// Whether the stream is considered stalled (no data received within timeout).
+	stalled bool
+}
+
+// streamEvent represents an event from a stream (entry received or stream closed).
+type streamEvent struct {
+	stream *mergerStream
+	entry  api.ServiceLogEntry
+	closed bool
+}
+
+// queuedEntry wraps a log entry with its source semaphore for release tracking.
+type queuedEntry struct {
+	entry     api.ServiceLogEntry
+	semaphore chan struct{}
+}
+
+// run is the main processing loop that merges all streams.
+func (m *LogMerger) run() {
+	defer close(m.output)
+
+	// Fan-in channel for stream events.
+	events := make(chan streamEvent)
+
+	// Start a reader goroutine for each stream to send entries to the events channel with flow control.
+	var wg sync.WaitGroup
+	for _, stream := range m.streams {
+		wg.Go(func() {
+			for entry := range stream.stream {
+				// Acquire semaphore slot before sending the entry to limit in-flight unprocessed entries per stream.
+				stream.semaphore <- struct{}{}
+				events <- streamEvent{stream: stream, entry: entry}
+			}
+
+			events <- streamEvent{stream: stream, closed: true}
+		})
+	}
+
+	// Close events channel when all readers finish.
+	go func() {
+		wg.Wait()
+		close(events)
+	}()
+
+	// Set up stall detection timer if enabled.
+	var stallCh <-chan time.Time
+	if m.stallTimeout > 0 && m.stallCheckInterval > 0 {
+		stallTicker := time.NewTicker(m.stallCheckInterval)
+		stallCh = stallTicker.C
+		defer stallTicker.Stop()
+	}
+
+	// Process events and emit entries.
+	for {
+		select {
+		case e, ok := <-events:
+			if !ok {
+				// All streams closed: flush remaining entries in order.
+				for m.queue.Len() > 0 {
+					qe := heap.Pop(&m.queue).(queuedEntry)
+					m.output <- qe.entry
+					<-qe.semaphore
+				}
+				return
+			}
+
+			e.stream.lastActivity = time.Now()
+			if e.stream.stalled {
+				e.stream.stalled = false
+			}
+			if e.stream.metadata == nil {
+				e.stream.metadata = &e.entry.Metadata
+			}
+
+			if e.closed {
+				e.stream.closed = true
+
+				m.updateWatermark()
+				m.emitReadyEntries()
+				continue
+			}
+
+			// Forward errors immediately and release semaphore.
+			if e.entry.Err != nil {
+				m.output <- e.entry
+				<-e.stream.semaphore
+				continue
+			}
+
+			if e.entry.Timestamp.After(e.stream.lastSeen) {
+				e.stream.lastSeen = e.entry.Timestamp
+			}
+			if e.entry.Stream == api.LogStreamStdout || e.entry.Stream == api.LogStreamStderr {
+				heap.Push(&m.queue, queuedEntry{entry: e.entry, semaphore: e.stream.semaphore})
+			}
+
+			m.updateWatermark()
+			m.emitReadyEntries()
+
+			// When merging streams, each input emits its own heartbeats. We want to debounce them and emit our own
+			// heartbeats at the same rate as a single input stream. Note that we need to adjust the heartbeat timestamp
+			// to the current watermark to not violate ordering guarantees.
+			if e.entry.Stream == api.LogStreamHeartbeat {
+				if m.watermark.Sub(m.lastEmitted) >= logMergerHeartbeatDebounceInterval {
+					heartbeat := e.entry
+					heartbeat.Timestamp = m.watermark
+					m.output <- heartbeat
+					m.lastEmitted = m.watermark
+				}
+				// Heartbeat processed, release semaphore.
+				<-e.stream.semaphore
+			}
+
+		case <-stallCh:
+			stalled := m.checkStalledStreams()
+			if len(stalled) == 0 {
+				continue
+			}
+
+			for _, s := range stalled {
+				errEntry := api.ServiceLogEntry{
+					ContainerLogEntry: api.ContainerLogEntry{
+						Err: api.ErrLogStreamStalled,
+					},
+				}
+				if s.metadata != nil {
+					errEntry.Metadata = *s.metadata
+				}
+
+				m.output <- errEntry
+			}
+
+			m.updateWatermark()
+			m.emitReadyEntries()
+		}
+	}
+}
+
+// checkStalledStreams marks streams as stalled if they haven't received any data within the timeout.
+// Returns true if any stream's stalled state changed.
+func (m *LogMerger) checkStalledStreams() []*mergerStream {
+	var stalled []*mergerStream
+	now := time.Now()
+
+	for _, s := range m.streams {
+		if s.closed || s.stalled {
+			continue
+		}
+
+		if now.Sub(s.lastActivity) > m.stallTimeout {
+			s.stalled = true
+			stalled = append(stalled, s)
+		}
+	}
+
+	return stalled
+}
+
+// updateWatermark recalculates the low watermark based on the lastSeen timestamps of all active streams.
+func (m *LogMerger) updateWatermark() {
+	first := true
+
+	for _, s := range m.streams {
+		if s.closed || s.stalled {
+			// Closed and stalled streams don't affect watermark.
+			continue
+		}
+		if first || s.lastSeen.Before(m.watermark) {
+			m.watermark = s.lastSeen
+			first = false
+		}
+	}
+}
+
+// emitReadyEntries pops and emits all buffered entries from the queue with timestamp before the watermark.
+func (m *LogMerger) emitReadyEntries() {
+	if m.watermark.IsZero() {
+		// No entries received yet.
+		return
+	}
+
+	for m.queue.Len() > 0 && m.queue[0].entry.Timestamp.Compare(m.watermark) <= 0 {
+		qe := heap.Pop(&m.queue).(queuedEntry)
+		m.output <- qe.entry
+		m.lastEmitted = qe.entry.Timestamp
+		<-qe.semaphore
+	}
+}
+
+// logsHeap is a min-heap (heap.Interface) of queued entries ordered by timestamp.
+type logsHeap []queuedEntry
+
+func (h *logsHeap) Len() int {
+	return len(*h)
+}
+
+func (h *logsHeap) Less(i, j int) bool {
+	return (*h)[i].entry.Timestamp.Before((*h)[j].entry.Timestamp)
+}
+
+func (h *logsHeap) Swap(i, j int) {
+	(*h)[i], (*h)[j] = (*h)[j], (*h)[i]
+}
+
+func (h *logsHeap) Push(x any) {
+	*h = append(*h, x.(queuedEntry))
+}
+
+func (h *logsHeap) Pop() any {
+	old := *h
+	n := len(old)
+	x := old[n-1]
+	*h = old[:n-1]
+	return x
+}

pkg/client/logmerger_test.go

@@ -0,0 +1,383 @@
+package client
+
+import (
+	"testing"
+	"time"
+
+	"github.com/psviderski/uncloud/pkg/api"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// testEntry creates a ServiceLogEntry for testing.
+func testEntry(stream api.LogStreamType, ts time.Time, msg string) api.ServiceLogEntry {
+	return api.ServiceLogEntry{
+		ContainerLogEntry: api.ContainerLogEntry{
+			Stream:    stream,
+			Timestamp: ts,
+			Message:   []byte(msg),
+		},
+	}
+}
+
+// collectEntries collects up to maxCount entries from the channel or until it is closed.
+func collectEntries(t *testing.T, ch <-chan api.ServiceLogEntry, maxCount int) []api.ServiceLogEntry {
+	t.Helper()
+	var entries []api.ServiceLogEntry
+
+	for i := 0; i < maxCount || maxCount <= 0; i++ {
+		select {
+		case e, ok := <-ch:
+			if !ok {
+				return entries
+			}
+			entries = append(entries, e)
+		case <-time.After(1 * time.Second):
+			require.FailNow(t, "timed out waiting for log entry")
+		}
+	}
+
+	return entries
+}
+
+func TestLogMerger_EmptyStreams(t *testing.T) {
+	t.Parallel()
+
+	merger := NewLogMerger(nil, LogMergerOptions{})
+	output := merger.Stream()
+
+	_, ok := <-output
+	assert.False(t, ok, "output channel should be closed for empty streams")
+}
+
+func TestLogMerger_SingleStream(t *testing.T) {
+	t.Parallel()
+
+	ch := make(chan api.ServiceLogEntry, 10)
+	merger := NewLogMerger([]<-chan api.ServiceLogEntry{ch}, LogMergerOptions{})
+	output := merger.Stream()
+
+	t1 := time.Now()
+	t2 := t1.Add(time.Second)
+
+	ch <- testEntry(api.LogStreamStdout, t1, "first")
+	ch <- testEntry(api.LogStreamStdout, t2, "second")
+
+	results := collectEntries(t, output, 2)
+
+	require.Len(t, results, 2)
+	assert.Equal(t, "first", string(results[0].Message))
+	assert.Equal(t, "second", string(results[1].Message))
+
+	close(ch)
+
+	results = collectEntries(t, output, 0)
+	assert.Len(t, results, 0, "no entries expected after close")
+}
+
+func TestLogMerger_PreservesData(t *testing.T) {
+	t.Parallel()
+
+	ch := make(chan api.ServiceLogEntry, 10)
+	merger := NewLogMerger([]<-chan api.ServiceLogEntry{ch}, LogMergerOptions{})
+	output := merger.Stream()
+
+	metadata := api.ServiceLogEntryMetadata{
+		ServiceID:   "svc-123",
+		ServiceName: "my-service",
+		MachineID:   "machine-456",
+		MachineName: "machine-1",
+	}
+
+	e := api.ServiceLogEntry{
+		Metadata: metadata,
+		ContainerLogEntry: api.ContainerLogEntry{
+			Stream:    api.LogStreamStdout,
+			Timestamp: time.Now(),
+			Message:   []byte("test"),
+		},
+	}
+	ch <- e
+	close(ch)
+
+	results := collectEntries(t, output, 0)
+	require.Len(t, results, 1)
+	assert.Equal(t, e, results[0])
+}
+
+func TestLogMerger_BasicMerge(t *testing.T) {
+	t.Parallel()
+
+	ch1 := make(chan api.ServiceLogEntry, 10)
+	ch2 := make(chan api.ServiceLogEntry, 10)
+
+	merger := NewLogMerger([]<-chan api.ServiceLogEntry{ch1, ch2}, LogMergerOptions{})
+	output := merger.Stream()
+
+	t1 := time.Now()
+	t2 := t1.Add(time.Second)
+	t3 := t1.Add(2 * time.Second)
+	t4 := t1.Add(3 * time.Second)
+
+	// Send interleaved entries from both streams.
+	ch1 <- testEntry(api.LogStreamStdout, t1, "ch1-first")
+	ch2 <- testEntry(api.LogStreamStdout, t2, "ch2-first")
+	ch1 <- testEntry(api.LogStreamStdout, t3, "ch1-second")
+	ch2 <- testEntry(api.LogStreamStdout, t4, "ch2-second")
+
+	// Send heartbeats to advance watermark past all entries.
+	t5 := t1.Add(4 * time.Second)
+	ch1 <- testEntry(api.LogStreamHeartbeat, t5, "")
+	ch2 <- testEntry(api.LogStreamHeartbeat, t5, "")
+
+	results := collectEntries(t, output, 4)
+	require.Len(t, results, 4)
+	assert.Equal(t, "ch1-first", string(results[0].Message))
+	assert.Equal(t, "ch2-first", string(results[1].Message))
+	assert.Equal(t, "ch1-second", string(results[2].Message))
+	assert.Equal(t, "ch2-second", string(results[3].Message))
+
+	close(ch1)
+	close(ch2)
+
+	results = collectEntries(t, output, 0)
+	assert.Len(t, results, 1, "one debounced heartbeat expected after close")
+	assert.Equal(t, api.LogStreamHeartbeat, results[0].Stream)
+	assert.Equal(t, t5, results[0].Timestamp)
+}
+
+func TestLogMerger_HeartbeatAdvancesWatermark(t *testing.T) {
+	t.Parallel()
+
+	ch1 := make(chan api.ServiceLogEntry, 10)
+	ch2 := make(chan api.ServiceLogEntry, 10)
+
+	merger := NewLogMerger([]<-chan api.ServiceLogEntry{ch1, ch2}, LogMergerOptions{})
+	output := merger.Stream()
+
+	t1 := time.Now()
+	t2 := t1.Add(time.Second)
+	t3 := t1.Add(2 * time.Second)
+
+	// Stream 1 sends a log entry.
+	ch1 <- testEntry(api.LogStreamStdout, t1, "ch1-first")
+	ch1 <- testEntry(api.LogStreamStdout, t3, "ch1-second")
+	// Stream 2 is quiet but sends a heartbeat.
+	ch2 <- testEntry(api.LogStreamHeartbeat, t2, "")
+
+	// Now stream 1's entry should be emitted because watermark is t1 (min of t1, t2).
+	results := collectEntries(t, output, 1)
+	require.Len(t, results, 1)
+	assert.Equal(t, "ch1-first", string(results[0].Message))
+
+	close(ch1)
+	close(ch2)
+
+	results = collectEntries(t, output, 0)
+	require.NotEmpty(t, results)
+
+	// Depending on timing, we may get a heartbeat before the second log entry.
+	require.LessOrEqual(t, len(results), 2, "one or two entries expected after close")
+	if len(results) == 2 {
+		assert.Equal(t, api.LogStreamHeartbeat, results[0].Stream)
+		assert.Equal(t, t2, results[0].Timestamp)
+		results = results[1:]
+	}
+	assert.Equal(t, "ch1-second", string(results[0].Message))
+}
+
+func TestLogMerger_ErrorForwarding(t *testing.T) {
+	t.Parallel()
+
+	ch1 := make(chan api.ServiceLogEntry, 10)
+	ch2 := make(chan api.ServiceLogEntry, 10)
+	merger := NewLogMerger([]<-chan api.ServiceLogEntry{ch1, ch2}, LogMergerOptions{})
+	output := merger.Stream()
+
+	t1 := time.Now()
+
+	ch1 <- testEntry(api.LogStreamStdout, t1, "ch1-first")
+	// Send an error entry.
+	ch1 <- api.ServiceLogEntry{
+		ContainerLogEntry: api.ContainerLogEntry{
+			Err: assert.AnError,
+		},
+	}
+
+	results := collectEntries(t, output, 1)
+	require.Len(t, results, 1, "expected error to emitted out of order")
+	assert.Equal(t, assert.AnError, results[0].Err)
+
+	close(ch1)
+	close(ch2)
+
+	results = collectEntries(t, output, 0)
+	assert.Len(t, results, 1)
+	assert.Equal(t, "ch1-first", string(results[0].Message))
+}
+
+func TestLogMerger_OutOfOrderSingleStream(t *testing.T) {
+	t.Parallel()
+
+	ch := make(chan api.ServiceLogEntry, 10)
+	merger := NewLogMerger([]<-chan api.ServiceLogEntry{ch}, LogMergerOptions{})
+	output := merger.Stream()
+
+	t1 := time.Now()
+	t2 := t1.Add(time.Second)
+	t3 := t1.Add(2 * time.Second)
+	t4 := t1.Add(3 * time.Second)
+
+	// Send entries out of order within the stream.
+	ch <- testEntry(api.LogStreamStdout, t3, "third")
+	ch <- testEntry(api.LogStreamStdout, t2, "second")
+	ch <- testEntry(api.LogStreamStdout, t4, "forth")
+	ch <- testEntry(api.LogStreamStdout, t1, "first")
+
+	results := collectEntries(t, output, 4)
+	require.Len(t, results, 4)
+	// Emitted in the same order as sent since no buffering/reordering is done within a single stream.
+	assert.Equal(t, "third", string(results[0].Message))
+	assert.Equal(t, "second", string(results[1].Message))
+	assert.Equal(t, "forth", string(results[2].Message))
+	assert.Equal(t, "first", string(results[3].Message))
+
+	close(ch)
+
+	results = collectEntries(t, output, 0)
+	assert.Len(t, results, 0, "no entries expected after close")
+}
+
+// Test that entries from streams with different rates are merged correctly in chronological order.
+func TestLogMerger_UnevenStreams(t *testing.T) {
+	t.Parallel()
+
+	numFastEntries := logMergerMaxInFlightPerStream
+
+	// Use buffered channels to allow sends without blocking on receiver.
+	ch1 := make(chan api.ServiceLogEntry, numFastEntries+5)
+	ch2 := make(chan api.ServiceLogEntry, 10)
+
+	merger := NewLogMerger([]<-chan api.ServiceLogEntry{ch1, ch2}, LogMergerOptions{})
+	output := merger.Stream()
+
+	baseTime := time.Now()
+
+	// Stream 1 sends many entries quickly (0ms - 104ms).
+	for i := 0; i < numFastEntries; i++ {
+		ch1 <- testEntry(api.LogStreamStdout, baseTime.Add(time.Duration(i)*time.Millisecond), "fast")
+	}
+
+	// Stream 2 sends entries that interleave with fast entries.
+	ch2 <- testEntry(api.LogStreamStdout, baseTime.Add(5*time.Millisecond), "slow")
+	ch2 <- testEntry(api.LogStreamStdout, baseTime.Add(50*time.Millisecond), "slow")
+
+	// Send heartbeats to advance watermark past all entries so they get emitted.
+	finalTime := baseTime.Add(time.Second)
+	ch1 <- testEntry(api.LogStreamHeartbeat, finalTime, "")
+	ch2 <- testEntry(api.LogStreamHeartbeat, finalTime.Add(10*time.Millisecond), "")
+
+	// Read exactly numFastEntries + 2 log entries.
+	expectedCount := numFastEntries + 2
+	results := collectEntries(t, output, expectedCount)
+	require.Len(t, results, expectedCount)
+
+	// Count entries from each stream.
+	fastCount := 0
+	slowCount := 0
+	for _, entry := range results {
+		switch string(entry.Message) {
+		case "fast":
+			fastCount++
+		case "slow":
+			slowCount++
+		}
+	}
+	assert.Equal(t, numFastEntries, fastCount)
+	assert.Equal(t, 2, slowCount)
+
+	// Verify entries are in chronological order.
+	for i := 1; i < len(results); i++ {
+		assert.False(t, results[i].Timestamp.Before(results[i-1].Timestamp),
+			"entry %d (ts=%v) should not be before entry %d (ts=%v)",
+			i, results[i].Timestamp, i-1, results[i-1].Timestamp)
+	}
+
+	close(ch1)
+	close(ch2)
+
+	results = collectEntries(t, output, 0)
+	require.Len(t, results, 1, "one debounced heartbeat expected after close")
+	assert.Equal(t, api.LogStreamHeartbeat, results[0].Stream)
+}
+
+func TestLogMerger_StalledStreamExcludedFromWatermark(t *testing.T) {
+	t.Parallel()
+
+	ch1 := make(chan api.ServiceLogEntry, 10)
+	ch2 := make(chan api.ServiceLogEntry, 10)
+
+	stallTimeout := 100 * time.Millisecond
+	merger := NewLogMerger([]<-chan api.ServiceLogEntry{ch1, ch2}, LogMergerOptions{
+		StallTimeout:       stallTimeout,
+		StallCheckInterval: 20 * time.Millisecond,
+	})
+	output := merger.Stream()
+
+	//// Collect all output entries in a separate goroutine.
+	//var results []api.ServiceLogEntry
+	//doneCollecting := make(chan struct{})
+	//go func() {
+	//	for entry := range output {
+	//		results = append(results, entry)
+	//	}
+	//	close(doneCollecting)
+	//}()
+
+	t1 := time.Now()
+	t2 := t1.Add(time.Second)
+	t3 := t1.Add(2 * time.Second)
+	t4 := t1.Add(3 * time.Second)
+
+	e1 := testEntry(api.LogStreamStdout, t1, "ch1-entry")
+	e1.Metadata = api.ServiceLogEntryMetadata{
+		ServiceID:   "ch1-svc1",
+		ServiceName: "ch1-svcName",
+		ContainerID: "ch1-ctr1",
+		MachineID:   "ch1-machine1",
+		MachineName: "ch1-machineName",
+	}
+
+	ch1 <- e1
+	ch2 <- testEntry(api.LogStreamStdout, t2, "ch2-first")
+
+	// Watermark is t1 now so ch1's entry could be collected.
+	results := collectEntries(t, output, 1)
+	require.Len(t, results, 1)
+	assert.Equal(t, "ch1-entry", string(results[0].Message))
+
+	// Keep pushing to stream 2 so that it does not stall. But it can't emit anything because watermark is stuck at t1.
+	time.Sleep(stallTimeout / 2)
+	ch2 <- testEntry(api.LogStreamStdout, t3, "ch2-second")
+
+	// 1/2 + 2/3 = 7/6 > 1, so stream 1 should be considered stalled now.
+	time.Sleep(stallTimeout / 3 * 2)
+	ch2 <- testEntry(api.LogStreamStdout, t4, "ch2-third")
+
+	results = collectEntries(t, output, 1)
+	require.ErrorIs(t, results[0].Err, api.ErrLogStreamStalled)
+	assert.Equal(t, e1.Metadata, results[0].Metadata, "stalled entry should have stream metadata")
+
+	// Now that stream 1 is marked as stalled and ignored, all in-flight entries from stream 2 are emitted.
+	results = collectEntries(t, output, 3)
+	require.Len(t, results, 3)
+	assert.Equal(t, "ch2-first", string(results[0].Message))
+	assert.Equal(t, "ch2-second", string(results[1].Message))
+	assert.Equal(t, "ch2-third", string(results[2].Message))
+
+	close(ch1)
+	close(ch2)
+
+	results = collectEntries(t, output, 0)
+	assert.Len(t, results, 0)
+}

pkg/client/logs.go

@@ -0,0 +1,141 @@
+package client
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/docker/docker/pkg/stringid"
+	"github.com/psviderski/uncloud/internal/machine/api/pb"
+	"github.com/psviderski/uncloud/pkg/api"
+)
+
+// ServiceLogs streams log entries from all service containers in chronological order based on timestamps.
+// Keep in mind that perfect ordering of log events across multiple machines can't be guaranteed due to the
+// imperfection of physical clocks or potential clock skew between machines.
+// It uses a low watermark algorithm to ensure proper ordering across multiple machines.
+// Heartbeat entries from the server advance the watermark to enable timely emission of buffered logs.
+func (cli *Client) ServiceLogs(
+	ctx context.Context, serviceNameOrID string, opts api.ServiceLogsOptions,
+) (api.Service, <-chan api.ServiceLogEntry, error) {
+	svc, err := cli.InspectService(ctx, serviceNameOrID)
+	if err != nil {
+		return svc, nil, fmt.Errorf("inspect service: %w", err)
+	}
+
+	if len(svc.Containers) == 0 {
+		return svc, nil, fmt.Errorf("no containers found for service: %s", serviceNameOrID)
+	}
+
+	machines, err := cli.ListMachines(ctx, nil)
+	if err != nil {
+		return svc, nil, fmt.Errorf("list machines: %w", err)
+	}
+
+	ctrStreams := make([]<-chan api.ServiceLogEntry, 0, len(svc.Containers))
+	for _, ctr := range svc.Containers {
+		// Try to get machine name for ServiceLogEntry metadata and friendlier error message.
+		machineName := ctr.MachineID
+		m := machines.FindByNameOrID(ctr.MachineID)
+		if m != nil {
+			machineName = m.Machine.Name
+		}
+
+		stream, err := cli.ContainerLogs(ctx, ctr.MachineID, ctr.Container.ID, opts)
+		if err != nil {
+			return svc, nil, fmt.Errorf("stream logs from service container '%s' on machine '%s': %w",
+				stringid.TruncateID(ctr.Container.ID), machineName, err)
+		}
+
+		// Enrich log entries from the container with service metadata.
+		metadata := api.ServiceLogEntryMetadata{
+			ServiceID:   svc.ID,
+			ServiceName: svc.Name,
+			ContainerID: ctr.Container.ID,
+			MachineID:   ctr.MachineID,
+			MachineName: machineName,
+		}
+		enrichedStream := logsStreamWithServiceMetadata(stream, metadata)
+		ctrStreams = append(ctrStreams, enrichedStream)
+	}
+
+	// Use the log merger to combine streams from all containers in chronological order.
+	merger := NewLogMerger(ctrStreams, DefaultLogMergerOptions)
+	mergedStream := merger.Stream()
+
+	return svc, mergedStream, nil
+}
+
+// ContainerLogs streams log entries from a single container on a specified machine.
+func (cli *Client) ContainerLogs(
+	ctx context.Context, machineNameOrID string, containerID string, opts api.ServiceLogsOptions,
+) (<-chan api.ContainerLogEntry, error) {
+	proxyCtx, _, err := cli.ProxyMachinesContext(ctx, []string{machineNameOrID})
+	if err != nil {
+		return nil, fmt.Errorf("create request context to proxy to machine '%s': %w", machineNameOrID, err)
+	}
+
+	req := &pb.ContainerLogsRequest{
+		ContainerId: containerID,
+		Follow:      opts.Follow,
+		Tail:        int32(opts.Tail),
+		Since:       opts.Since,
+		Until:       opts.Until,
+	}
+	stream, err := cli.Docker.GRPCClient.ContainerLogs(proxyCtx, req)
+	if err != nil {
+		return nil, err
+	}
+
+	ch := make(chan api.ContainerLogEntry)
+
+	go func() {
+		defer close(ch)
+
+		for {
+			pbEntry, err := stream.Recv()
+			if err == io.EOF {
+				return
+			}
+			if err != nil {
+				ch <- api.ContainerLogEntry{
+					Err: err,
+				}
+				return
+			}
+
+			entry := api.ContainerLogEntry{
+				Stream:    api.LogStreamTypeFromProto(pbEntry.Stream),
+				Message:   pbEntry.Message,
+				Timestamp: pbEntry.Timestamp.AsTime(),
+			}
+
+			select {
+			case ch <- entry:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return ch, nil
+}
+
+// logsStreamWithServiceMetadata wraps a container logs stream and enriches each log entry with service metadata.
+func logsStreamWithServiceMetadata(
+	stream <-chan api.ContainerLogEntry, metadata api.ServiceLogEntryMetadata,
+) <-chan api.ServiceLogEntry {
+	out := make(chan api.ServiceLogEntry)
+
+	go func() {
+		for entry := range stream {
+			out <- api.ServiceLogEntry{
+				Metadata:          metadata,
+				ContainerLogEntry: entry,
+			}
+		}
+		close(out)
+	}()
+
+	return out
+}

pkg/client/machine.go

@@ -48,6 +48,26 @@ func (cli *Client) ListMachines(ctx context.Context, filter *api.MachineFilter)
 	return machines, nil
 }
 
+func MachineMatchesFilter(machine *pb.MachineMember, filter *api.MachineFilter) bool {
+	if filter == nil {
+		return true
+	}
+
+	if filter.Available && machine.State == pb.MachineMember_DOWN {
+		return false
+	}
+
+	if len(filter.NamesOrIDs) > 0 {
+		if !slices.ContainsFunc(filter.NamesOrIDs, func(nameOrID string) bool {
+			return machine.Machine.Id == nameOrID || machine.Machine.Name == nameOrID
+		}) {
+			return false
+		}
+	}
+
+	return true
+}
+
 // UpdateMachine updates machine configuration in the cluster.
 func (cli *Client) UpdateMachine(ctx context.Context, req *pb.UpdateMachineRequest) (*pb.MachineInfo, error) {
 	resp, err := cli.ClusterClient.UpdateMachine(ctx, req)
@@ -76,23 +96,3 @@ func (cli *Client) RenameMachine(ctx context.Context, nameOrID, newName string)
 
 	return cli.UpdateMachine(ctx, req)
 }
-
-func MachineMatchesFilter(machine *pb.MachineMember, filter *api.MachineFilter) bool {
-	if filter == nil {
-		return true
-	}
-
-	if filter.Available && machine.State == pb.MachineMember_DOWN {
-		return false
-	}
-
-	if len(filter.NamesOrIDs) > 0 {
-		if !slices.ContainsFunc(filter.NamesOrIDs, func(nameOrID string) bool {
-			return machine.Machine.Id == nameOrID || machine.Machine.Name == nameOrID
-		}) {
-			return false
-		}
-	}
-
-	return true
-}

pkg/client/volume.go

@@ -55,7 +55,7 @@ func (cli *Client) ListVolumes(ctx context.Context, filter *api.VolumeFilter) ([
 		proxyMachines = filter.Machines
 	}
 
-	listCtx, machines, err := api.ProxyMachinesContext(ctx, cli, proxyMachines)
+	listCtx, machines, err := cli.ProxyMachinesContext(ctx, proxyMachines)
 	if err != nil {
 		return nil, fmt.Errorf("create request context to broadcast to all machines: %w", err)
 	}