feat(cloud): proxy alert history to the local web layer (#4914)

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

internal/cloud/alerts.go

@@ -0,0 +1,117 @@
+package cloud
+
+import (
+	"context"
+	"fmt"
+
+	pb "github.com/amir20/dozzle/proto/cloud"
+	"google.golang.org/grpc/metadata"
+)
+
+// AlertResult is the JSON-friendly response shape returned to the Dozzle web
+// layer. Mirrors the proto GetAlertsResponse but lives in this package so
+// callers don't have to import the proto package directly.
+type AlertResult struct {
+	Hits []AlertHit `json:"hits"`
+	// Truncated means the window held more anchors than the limit allowed, so
+	// the viewer is not seeing all of them.
+	Truncated bool `json:"truncated,omitempty"`
+}
+
+// AlertHit is one alert anchored inside the requested window, scoped
+// server-side to the connecting instance's (user_id, api_key_id) — Cloud
+// derives those from the auth metadata, never the request body.
+type AlertHit struct {
+	AlertID     int64  `json:"alertId"`
+	ContainerID string `json:"containerId"`
+	HostID      string `json:"hostId"`
+	// LogID is Cloud's copy of Dozzle's FNV-32a hash of the line that
+	// triggered the alert — the same id LogEvent.Id carries, so the viewer can
+	// splice the alert in directly after that line. Omitted (0) for metric and
+	// event alerts, which have no triggering line and anchor on Ts instead.
+	LogID uint32 `json:"logId,omitempty"`
+	Ts    int64  `json:"ts"`
+
+	Headline        string `json:"headline"`
+	Level           string `json:"level"`
+	EventCount      int32  `json:"eventCount"`
+	SuppressedCount int32  `json:"suppressedCount,omitempty"`
+	// ContainerCount > 1 means the incident is wider than the container being
+	// viewed.
+	ContainerCount int32  `json:"containerCount,omitempty"`
+	Investigation  string `json:"investigation,omitempty"`
+	TriageAction   string `json:"triageAction,omitempty"`
+
+	CreatedAt      int64 `json:"createdAt"`
+	LastActivityAt int64 `json:"lastActivityAt,omitempty"`
+	// IsOrigin marks the anchor where the incident FIRST fired. Cloud appends
+	// every folded batch's events to the original alert, so one incident can
+	// have activity in many scroll windows; the viewer renders a full card at
+	// the origin and a compact "still happening" marker at follow-ups.
+	IsOrigin bool `json:"isOrigin"`
+
+	// URL deep-links to the alert page in Dozzle Cloud. Empty when the API key
+	// has no app_url configured.
+	URL string `json:"url,omitempty"`
+}
+
+// GetAlerts fetches the alerts that fired on these containers inside a window,
+// for merging into the local log stream on scrollback. Reuses the long-lived
+// unary conn so a scroll doesn't pay a TLS handshake.
+//
+// Identity (user, instance) is enforced server-side from the authenticated
+// metadata; this client passes only the per-request fields below. Unlike
+// SearchLogs this does not depend on the streamLogs opt-in — alerts live in
+// Cloud's own database rather than the log store.
+func (c *Client) GetAlerts(ctx context.Context, containerIDs []string, hostID string, fromNs, toNs int64, limit int32, includeFollowUps bool) (*AlertResult, error) {
+	apiKey := c.apiKeyFunc()
+	if apiKey == "" {
+		return nil, ErrNotConfigured
+	}
+
+	client, err := c.unaryServiceClient()
+	if err != nil {
+		return nil, err
+	}
+
+	mdPairs := []string{"x-api-key", apiKey}
+	if c.instanceID != "" {
+		mdPairs = append(mdPairs, "x-instance-id", c.instanceID)
+	}
+	callCtx := metadata.NewOutgoingContext(ctx, metadata.Pairs(mdPairs...))
+
+	resp, err := client.GetAlerts(callCtx, &pb.GetAlertsRequest{
+		ContainerIds:     containerIDs,
+		HostId:           hostID,
+		FromTsNs:         fromNs,
+		ToTsNs:           toNs,
+		Limit:            limit,
+		IncludeFollowUps: includeFollowUps,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("cloud: alerts: %w", err)
+	}
+
+	hits := make([]AlertHit, 0, len(resp.GetHits()))
+	for _, h := range resp.GetHits() {
+		hits = append(hits, AlertHit{
+			AlertID:         h.GetAlertId(),
+			ContainerID:     h.GetContainerId(),
+			HostID:          h.GetHostId(),
+			LogID:           h.GetLogId(),
+			Ts:              h.GetAnchorTsNs(),
+			Headline:        h.GetHeadline(),
+			Level:           h.GetLevel(),
+			EventCount:      h.GetEventCount(),
+			SuppressedCount: h.GetSuppressedCount(),
+			ContainerCount:  h.GetContainerCount(),
+			Investigation:   h.GetInvestigation(),
+			TriageAction:    h.GetTriageAction(),
+			CreatedAt:       h.GetCreatedAtNs(),
+			LastActivityAt:  h.GetLastActivityAtNs(),
+			IsOrigin:        h.GetIsOrigin(),
+			URL:             h.GetUrl(),
+		})
+	}
+	return &AlertResult{Hits: hits, Truncated: resp.GetTruncated()}, nil
+}

internal/cloud/client.go

@@ -52,13 +52,14 @@ type Client struct {
 	connMu        sync.Mutex
 	cancelCurrent context.CancelFunc
 
-	// searchConn / searchClient are lazily initialized and shared across
-	// SearchLogs calls so we don't pay the TLS handshake on every keystroke.
-	// Same target / TLS as the main ToolStream conn; per-call identity is
-	// supplied via metadata (x-api-key, x-instance-id), so one conn is fine.
-	searchConnMu sync.Mutex
-	searchConn   *grpc.ClientConn
-	searchClient pb.CloudToolServiceClient
+	// unaryConn / unaryClient are lazily initialized and shared across every
+	// Dozzle-initiated unary call (SearchLogs, GetAlerts) so we don't pay the
+	// TLS handshake per keystroke or per scroll. Same target / TLS as the main
+	// ToolStream conn; per-call identity is supplied via metadata (x-api-key,
+	// x-instance-id), so one conn serves all of them.
+	unaryConnMu sync.Mutex
+	unaryConn   *grpc.ClientConn
+	unaryClient pb.CloudToolServiceClient
 }
 
 // NewClient creates a new cloud gRPC client.

internal/cloud/search.go

@@ -45,14 +45,14 @@ type SearchLogHit struct {
 // is available (the user hasn't linked Cloud yet). Callers map this to a 503.
 var ErrNotConfigured = errors.New("cloud: no API key configured")
 
-// searchServiceClient returns a (lazily dialed) reusable gRPC client. The
-// underlying conn is shared across all SearchLogs calls so we pay the TLS
-// handshake once per process — not once per keystroke.
-func (c *Client) searchServiceClient() (pb.CloudToolServiceClient, error) {
-	c.searchConnMu.Lock()
-	defer c.searchConnMu.Unlock()
-	if c.searchClient != nil {
-		return c.searchClient, nil
+// unaryServiceClient returns a (lazily dialed) reusable gRPC client. The
+// underlying conn is shared across every Dozzle-initiated unary call so we pay
+// the TLS handshake once per process — not once per keystroke or scroll.
+func (c *Client) unaryServiceClient() (pb.CloudToolServiceClient, error) {
+	c.unaryConnMu.Lock()
+	defer c.unaryConnMu.Unlock()
+	if c.unaryClient != nil {
+		return c.unaryClient, nil
 	}
 	var creds grpc.DialOption
 	if c.plaintext {
@@ -64,9 +64,9 @@ func (c *Client) searchServiceClient() (pb.CloudToolServiceClient, error) {
 	if err != nil {
 		return nil, fmt.Errorf("cloud: dial: %w", err)
 	}
-	c.searchConn = conn
-	c.searchClient = pb.NewCloudToolServiceClient(conn)
-	return c.searchClient, nil
+	c.unaryConn = conn
+	c.unaryClient = pb.NewCloudToolServiceClient(conn)
+	return c.unaryClient, nil
 }
 
 // SearchLogs runs a Cloud-side log search against the existing gRPC service.
@@ -80,7 +80,7 @@ func (c *Client) SearchLogs(ctx context.Context, query string, limit int32, host
 		return nil, ErrNotConfigured
 	}
 
-	client, err := c.searchServiceClient()
+	client, err := c.unaryServiceClient()
 	if err != nil {
 		return nil, err
 	}

internal/web/cloud_alerts.go

@@ -0,0 +1,121 @@
+package web
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/amir20/dozzle/internal/cloud"
+	"github.com/rs/zerolog/log"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// cloudAlertsTimeout caps the round-trip to Doligence Cloud. Shorter than the
+// search timeout: this rides the scroll path, where the log lines themselves
+// have already loaded and the alerts are a decoration on top. Better to render
+// the logs without alerts than to hold the scroll waiting for them.
+const cloudAlertsTimeout = 1500 * time.Millisecond
+
+// maxAlertContainers mirrors Cloud's own cap. The viewer sends the containers
+// currently on screen, which is a handful even in a merged multi-container
+// view, so anything larger is a misbehaving client.
+const maxAlertContainers = 100
+
+// cloudAlerts proxies an alert-history lookup to Doligence Cloud over the
+// existing authenticated gRPC connection, so the log viewer can merge alerts
+// into the stream as the user scrolls back. Identity is derived server-side
+// from the API key — this handler passes neither user nor instance ids.
+//
+// Deliberately NOT gated on streamLogs, unlike cloudSearchLogs: alerts live in
+// Cloud's own database rather than the log store, so they exist for anyone who
+// linked cloud and configured a subscription. Gating them behind the
+// log-streaming opt-in would hide the feature from most users who'd benefit.
+//
+// Status mapping:
+//
+//	200 — hits returned (may be empty)
+//	400 — malformed window or an oversized container list
+//	503 — cloud not configured (no API key) or no GetAlerts func wired
+//	504 — cloud round-trip exceeded the timeout
+//	502 — any other cloud-side error
+func (h *handler) cloudAlerts(w http.ResponseWriter, r *http.Request) {
+	if h.config.Cloud.GetAlerts == nil {
+		writeError(w, http.StatusServiceUnavailable, "cloud not configured")
+		return
+	}
+
+	q := r.URL.Query()
+
+	var containerIDs []string
+	for id := range strings.SplitSeq(q.Get("containerIds"), ",") {
+		if id = strings.TrimSpace(id); id != "" {
+			containerIDs = append(containerIDs, id)
+		}
+	}
+	if len(containerIDs) == 0 {
+		// Nothing to look up. Answer with an empty result rather than an error
+		// so the caller's merge path stays uniform.
+		writeAlerts(w, &cloud.AlertResult{Hits: []cloud.AlertHit{}})
+		return
+	}
+	if len(containerIDs) > maxAlertContainers {
+		writeError(w, http.StatusBadRequest, "too many containerIds")
+		return
+	}
+
+	from, err := strconv.ParseInt(q.Get("from"), 10, 64)
+	if err != nil || from <= 0 {
+		writeError(w, http.StatusBadRequest, "missing or invalid from")
+		return
+	}
+	to, err := strconv.ParseInt(q.Get("to"), 10, 64)
+	if err != nil || to <= from {
+		writeError(w, http.StatusBadRequest, "missing or invalid to")
+		return
+	}
+
+	// Cloud caps server-side too; mirroring it here keeps a misbehaving client
+	// from tying up the scroll path with an oversized request.
+	limit := int32(50)
+	if v := q.Get("limit"); v != "" {
+		if n, err := strconv.ParseInt(v, 10, 32); err == nil && n > 0 {
+			if n > 200 {
+				n = 200
+			}
+			limit = int32(n)
+		}
+	}
+
+	ctx, cancel := context.WithTimeout(r.Context(), cloudAlertsTimeout)
+	defer cancel()
+
+	result, err := h.config.Cloud.GetAlerts(ctx, containerIDs, q.Get("hostId"), from, to, limit, q.Get("followUps") == "1")
+	if err != nil {
+		if errors.Is(err, cloud.ErrNotConfigured) {
+			writeError(w, http.StatusServiceUnavailable, "cloud not configured")
+			return
+		}
+		// A blown deadline surfaces two ways: context.DeadlineExceeded when our
+		// own ctx fires first, or a gRPC status when the server-side deadline
+		// trips. status.Code walks the %w wrap chain, so both map to a 504.
+		if errors.Is(err, context.DeadlineExceeded) || status.Code(err) == codes.DeadlineExceeded {
+			writeError(w, http.StatusGatewayTimeout, "cloud alerts timed out")
+			return
+		}
+		log.Warn().Err(err).Msg("cloud alerts failed")
+		writeError(w, http.StatusBadGateway, "cloud alerts failed")
+		return
+	}
+
+	writeAlerts(w, result)
+}
+
+func writeAlerts(w http.ResponseWriter, result *cloud.AlertResult) {
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(result)
+}

internal/web/cloud_alerts_test.go

@@ -0,0 +1,139 @@
+package web
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/amir20/dozzle/internal/cloud"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+type alertCall struct {
+	containerIDs     []string
+	hostID           string
+	from, to         int64
+	limit            int32
+	includeFollowUps bool
+}
+
+// alertHandler builds a handler wired to a stub GetAlerts, capturing what the
+// HTTP layer forwarded. A nil fn leaves the hook unset, which is how "cloud not
+// wired" is expressed.
+func alertHandler(t *testing.T, fn func(*alertCall) (*cloud.AlertResult, error)) (*handler, *alertCall) {
+	t.Helper()
+	got := &alertCall{}
+	h := &handler{config: &Config{}}
+	if fn != nil {
+		h.config.Cloud.GetAlerts = func(ctx context.Context, containerIDs []string, hostID string, fromNs, toNs int64, limit int32, includeFollowUps bool) (*cloud.AlertResult, error) {
+			*got = alertCall{containerIDs, hostID, fromNs, toNs, limit, includeFollowUps}
+			return fn(got)
+		}
+	}
+	return h, got
+}
+
+func doAlerts(h *handler, query string) *httptest.ResponseRecorder {
+	rr := httptest.NewRecorder()
+	h.cloudAlerts(rr, httptest.NewRequest(http.MethodGet, "/api/cloud/alerts?"+query, nil))
+	return rr
+}
+
+func okResult(*alertCall) (*cloud.AlertResult, error) {
+	return &cloud.AlertResult{Hits: []cloud.AlertHit{{AlertID: 1, ContainerID: "abc"}}}, nil
+}
+
+func TestCloudAlerts_UnwiredIs503(t *testing.T) {
+	h, _ := alertHandler(t, nil)
+	rr := doAlerts(h, "containerIds=abc&from=1&to=2")
+	require.Equal(t, http.StatusServiceUnavailable, rr.Code)
+}
+
+// Alerts must NOT be gated on the streamLogs opt-in the way log search is:
+// they live in Cloud's database rather than the log store. hostService is nil
+// here, so if the handler ever reached for CloudConfig() this would panic.
+func TestCloudAlerts_DoesNotRequireStreamLogs(t *testing.T) {
+	h, _ := alertHandler(t, okResult)
+	rr := doAlerts(h, "containerIds=abc&from=1&to=2")
+	require.Equal(t, http.StatusOK, rr.Code)
+}
+
+func TestCloudAlerts_ForwardsParams(t *testing.T) {
+	h, got := alertHandler(t, okResult)
+	rr := doAlerts(h, "containerIds=abc,%20def&hostId=h1&from=100&to=200&limit=7&followUps=1")
+
+	require.Equal(t, http.StatusOK, rr.Code)
+	require.Equal(t, []string{"abc", "def"}, got.containerIDs, "ids should be split and trimmed")
+	require.Equal(t, "h1", got.hostID)
+	require.Equal(t, int64(100), got.from)
+	require.Equal(t, int64(200), got.to)
+	require.Equal(t, int32(7), got.limit)
+	require.True(t, got.includeFollowUps)
+}
+
+// An empty container list is a no-op, not an error: the caller's merge path
+// stays uniform and cloud is never asked a pointless question.
+func TestCloudAlerts_EmptyContainersReturnsEmpty(t *testing.T) {
+	called := false
+	h, _ := alertHandler(t, func(*alertCall) (*cloud.AlertResult, error) {
+		called = true
+		return &cloud.AlertResult{}, nil
+	})
+
+	rr := doAlerts(h, "containerIds=&from=1&to=2")
+	require.Equal(t, http.StatusOK, rr.Code)
+	require.False(t, called, "cloud should not be queried for an empty container list")
+
+	var result cloud.AlertResult
+	require.NoError(t, json.Unmarshal(rr.Body.Bytes(), &result))
+	require.Empty(t, result.Hits)
+}
+
+func TestCloudAlerts_RejectsBadWindow(t *testing.T) {
+	for name, query := range map[string]string{
+		"missing from":   "containerIds=abc&to=2",
+		"missing to":     "containerIds=abc&from=1",
+		"to before from": "containerIds=abc&from=5&to=2",
+		"to equals from": "containerIds=abc&from=5&to=5",
+		"unparseable":    "containerIds=abc&from=nope&to=2",
+	} {
+		t.Run(name, func(t *testing.T) {
+			h, _ := alertHandler(t, okResult)
+			require.Equal(t, http.StatusBadRequest, doAlerts(h, query).Code)
+		})
+	}
+}
+
+func TestCloudAlerts_CapsLimitAndContainers(t *testing.T) {
+	h, got := alertHandler(t, okResult)
+	require.Equal(t, http.StatusOK, doAlerts(h, "containerIds=abc&from=1&to=2&limit=9999").Code)
+	require.Equal(t, int32(200), got.limit, "limit should be capped, mirroring cloud")
+
+	var ids strings.Builder
+	for i := 0; i <= maxAlertContainers; i++ {
+		ids.WriteString("c,")
+	}
+	require.Equal(t, http.StatusBadRequest, doAlerts(h, "containerIds="+ids.String()+"&from=1&to=2").Code)
+}
+
+func TestCloudAlerts_ErrorMapping(t *testing.T) {
+	for name, tc := range map[string]struct {
+		err  error
+		want int
+	}{
+		"not configured": {cloud.ErrNotConfigured, http.StatusServiceUnavailable},
+		"deadline":       {status.Error(codes.DeadlineExceeded, "too slow"), http.StatusGatewayTimeout},
+		"ctx deadline":   {context.DeadlineExceeded, http.StatusGatewayTimeout},
+		"other":          {status.Error(codes.Internal, "boom"), http.StatusBadGateway},
+	} {
+		t.Run(name, func(t *testing.T) {
+			h, _ := alertHandler(t, func(*alertCall) (*cloud.AlertResult, error) { return nil, tc.err })
+			require.Equal(t, tc.want, doAlerts(h, "containerIds=abc&from=1&to=2").Code)
+		})
+	}
+}

internal/web/routes.go

@@ -70,6 +70,10 @@ type CloudHooks struct {
 	// SearchLogs proxies a substring/word-filter query to Doligence Cloud
 	// over the authenticated gRPC connection. Nil when cloud is not wired.
 	SearchLogs func(ctx context.Context, query string, limit int32, hostID, containerID string, before int64) (*cloud.SearchLogResult, error)
+	// GetAlerts fetches the alerts that fired on a set of containers inside a
+	// time window, so the log viewer can merge them into the stream on
+	// scrollback. Nil when cloud is not wired.
+	GetAlerts func(ctx context.Context, containerIDs []string, hostID string, fromNs, toNs int64, limit int32, includeFollowUps bool) (*cloud.AlertResult, error)
 }
 
 type Authorization struct {
@@ -212,6 +216,7 @@ func createRouter(h *handler) *chi.Mux {
 				// Cloud API
 				r.Get("/cloud/status", h.cloudStatus)
 				r.Get("/cloud/search/logs", h.cloudSearchLogs)
+				r.Get("/cloud/alerts", h.cloudAlerts)
 				r.Get("/cloud/config", h.cloudConfig)
 				r.Patch("/cloud/config", h.updateCloudConfig)
 				r.Delete("/cloud/config", h.deleteCloudConfig)

main.go

@@ -175,6 +175,7 @@ func main() {
 		OnSetup:    cloudClient.Notify,
 		OnUpdate:   cloudClient.Reconnect,
 		SearchLogs: cloudClient.SearchLogs,
+		GetAlerts:  cloudClient.GetAlerts,
 	})
 
 	go func() {