diff --git a/assets/components.d.ts b/assets/components.d.ts index 54787a12..a53870e1 100644 --- a/assets/components.d.ts +++ b/assets/components.d.ts @@ -131,6 +131,7 @@ declare module 'vue' { 'Mdi:poll': typeof import('~icons/mdi/poll')['default'] 'Mdi:refresh': typeof import('~icons/mdi/refresh')['default'] 'Mdi:satelliteVariant': typeof import('~icons/mdi/satellite-variant')['default'] + 'Mdi:shieldLockOutline': typeof import('~icons/mdi/shield-lock-outline')['default'] 'Mdi:textBoxOutline': typeof import('~icons/mdi/text-box-outline')['default'] 'Mdi:trashCanOutline': typeof import('~icons/mdi/trash-can-outline')['default'] 'Mdi:webhook': typeof import('~icons/mdi/webhook')['default'] diff --git a/assets/components/CloudSettingsCard.vue b/assets/components/CloudSettingsCard.vue index d91afa25..4bdc7df5 100644 --- a/assets/components/CloudSettingsCard.vue +++ b/assets/components/CloudSettingsCard.vue @@ -75,6 +75,25 @@ > + +
{{ $t("cloud.dashboard") }} @@ -125,6 +144,33 @@ const { const isUnlinking = ref(false); const unlinkModal = ref(null); +const streamLogs = ref(true); +const isSavingStreamLogs = ref(false); +watchEffect(() => { + if (cloudConfig.value) streamLogs.value = cloudConfig.value.streamLogs; +}); + +async function onStreamLogsChange(value: boolean | undefined) { + if (!cloudConfig.value || value === undefined) return; + isSavingStreamLogs.value = true; + try { + const res = await fetch(withBase("/api/cloud/config"), { + method: "PATCH", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ streamLogs: value }), + }); + if (!res.ok) { + streamLogs.value = !value; + return; + } + cloudConfig.value.streamLogs = value; + } catch { + streamLogs.value = !value; + } finally { + isSavingStreamLogs.value = false; + } +} + const usagePercent = computed(() => { if (!cloudStatus.value) return 0; return (cloudStatus.value.usage.events_used / cloudStatus.value.usage.events_limit) * 100; diff --git a/assets/types/notifications.ts b/assets/types/notifications.ts index 4abbaa25..8e4b4d66 100644 --- a/assets/types/notifications.ts +++ b/assets/types/notifications.ts @@ -71,6 +71,7 @@ export interface CloudConfig { prefix: string; expiresAt?: string; linked: boolean; + streamLogs: boolean; } export interface CloudStatus { diff --git a/internal/cloud/client.go b/internal/cloud/client.go index e1843001..7acb5a4f 100644 --- a/internal/cloud/client.go +++ b/internal/cloud/client.go @@ -36,16 +36,20 @@ const ( // Client manages the gRPC connection to Dozzle Cloud type Client struct { - deps ToolDeps - apiKeyFunc func() string - target string - plaintext bool - toolSem *semaphore.Weighted - streamSem *semaphore.Weighted - cachedTools []*pb.ToolDefinition - toolsOnce sync.Once - startCh chan struct{} - activeStreams sync.Map // requestID -> context.CancelFunc + deps ToolDeps + apiKeyFunc func() string + streamLogsFunc func() bool + target string + plaintext bool + toolSem *semaphore.Weighted + streamSem *semaphore.Weighted + cachedTools []*pb.ToolDefinition + toolsOnce sync.Once + startCh chan struct{} + activeStreams sync.Map // requestID -> context.CancelFunc + + connMu sync.Mutex + cancelCurrent context.CancelFunc } // NewClient creates a new cloud gRPC client. @@ -82,6 +86,12 @@ func NewClient(apiKeyFunc func() string, deps ToolDeps) *Client { } } +// SetStreamLogsFunc registers a function that reports whether bulk container +// log streaming to cloud is enabled. If unset, the streamer runs by default. +func (c *Client) SetStreamLogsFunc(f func() bool) { + c.streamLogsFunc = f +} + // Notify signals the client to attempt a connection. Safe to call multiple times. // Use this when a cloud dispatcher is added or when the status page is viewed. func (c *Client) Notify() { @@ -91,6 +101,17 @@ func (c *Client) Notify() { } } +// Reconnect drops the current cloud connection (if any), causing the Run loop +// to dial again so settings like the log-streaming toggle take effect. +func (c *Client) Reconnect() { + c.connMu.Lock() + cancel := c.cancelCurrent + c.connMu.Unlock() + if cancel != nil { + cancel() + } +} + // Run blocks until signaled via Notify(), then connects to the cloud gRPC endpoint // and processes tool requests. Reconnects automatically on failure. // Does nothing until Notify() is called — zero overhead for non-cloud users. @@ -192,8 +213,20 @@ func (c *Client) connect(ctx context.Context, apiKey string) (wasConnected bool, client := pb.NewCloudToolServiceClient(conn) + // Per-connection context that Reconnect() can cancel to force a redial. + connCtx, connCancel := context.WithCancel(ctx) + defer connCancel() + c.connMu.Lock() + c.cancelCurrent = connCancel + c.connMu.Unlock() + defer func() { + c.connMu.Lock() + c.cancelCurrent = nil + c.connMu.Unlock() + }() + md := metadata.Pairs("x-api-key", apiKey) - streamCtx := metadata.NewOutgoingContext(ctx, md) + streamCtx := metadata.NewOutgoingContext(connCtx, md) stream, err := client.ToolStream(streamCtx) if err != nil { @@ -212,6 +245,22 @@ func (c *Client) connect(ctx context.Context, apiKey string) (wasConnected bool, return stream.Send(resp) } + // Start the background log streamer if the host service supports it and + // the user has not opted out via the privacy toggle. Its lifetime is bound + // to streamLifetime — it shuts down cleanly when this connection drops, + // and is re-created on reconnect (re-evaluating the toggle each time). + streamLogs := c.streamLogsFunc == nil || c.streamLogsFunc() + if !streamLogs { + log.Debug().Msg("cloud log streaming disabled by user setting; skipping streamer") + } else if lshs, ok := c.deps.HostService.(LogStreamHostService); ok { + streamer := newLogStreamer(lshs, c.deps.Labels, sendResp) + wg.Go(func() { + streamer.run(streamLifetime) + }) + } else { + log.Debug().Msg("host service does not support log streaming; skipping") + } + defer func() { // Cancel all active log streams before shutting down c.activeStreams.Range(func(key, value any) bool { diff --git a/internal/cloud/log_streamer.go b/internal/cloud/log_streamer.go new file mode 100644 index 00000000..3fd02dfa --- /dev/null +++ b/internal/cloud/log_streamer.go @@ -0,0 +1,240 @@ +package cloud + +import ( + "context" + "encoding/json" + "sync" + "time" + + "github.com/amir20/dozzle/internal/container" + container_support "github.com/amir20/dozzle/internal/support/container" + pb "github.com/amir20/dozzle/proto/cloud" + "github.com/rs/zerolog/log" +) + +// LogStreamHostService is the subset of the host service needed by the log +// streamer. MultiHostService and K8sClusterService both satisfy it. +type LogStreamHostService interface { + ToolHostService + SubscribeContainersStarted(ctx context.Context, containers chan<- container.Container, filter container_support.ContainerFilter) +} + +const ( + logBatchMaxEntries = 500 + logBatchMaxBytes = 256 * 1024 + logBatchFlushPeriod = 1 * time.Second + logReaderChanBuffer = 128 +) + +// logStreamer streams raw container log lines to Dozzle Cloud as unsolicited +// LogBatch ToolResponses. It is created per cloud connection and torn down +// when the connection drops; a new one is started fresh on reconnect. +type logStreamer struct { + hostService LogStreamHostService + labels container.ContainerLabels + send func(resp *pb.ToolResponse) error + + mu sync.Mutex + readers map[string]context.CancelFunc + + wg sync.WaitGroup +} + +func newLogStreamer(hostService LogStreamHostService, labels container.ContainerLabels, send func(resp *pb.ToolResponse) error) *logStreamer { + return &logStreamer{ + hostService: hostService, + labels: labels, + send: send, + readers: make(map[string]context.CancelFunc), + } +} + +// run blocks until ctx is cancelled. It launches readers for all currently +// running containers and subscribes to new-container events to launch readers +// for containers started after connect. +func (ls *logStreamer) run(ctx context.Context) { + // Subscribe BEFORE snapshotting so we don't miss a container that starts + // between snapshot and subscribe. + started := make(chan container.Container, 64) + ls.hostService.SubscribeContainersStarted(ctx, started, func(_ *container.Container) bool { return true }) + + existing, errs := ls.hostService.ListAllContainers(ls.labels) + for _, err := range errs { + if err != nil { + log.Debug().Err(err).Msg("log streamer: error listing containers from host") + } + } + for _, c := range existing { + if c.State != "running" { + continue + } + ls.startReader(ctx, c) + } + + for { + select { + case <-ctx.Done(): + ls.wg.Wait() + return + case c, ok := <-started: + if !ok { + ls.wg.Wait() + return + } + if c.State != "running" { + continue + } + ls.startReader(ctx, c) + } + } +} + +func readerKey(hostID, containerID string) string { + return hostID + "|" + containerID +} + +func (ls *logStreamer) startReader(parent context.Context, c container.Container) { + key := readerKey(c.Host, c.ID) + + ls.mu.Lock() + if _, exists := ls.readers[key]; exists { + ls.mu.Unlock() + return + } + readerCtx, cancel := context.WithCancel(parent) + ls.readers[key] = cancel + ls.mu.Unlock() + + cs, err := ls.hostService.FindContainer(c.Host, c.ID, ls.labels) + if err != nil { + ls.mu.Lock() + delete(ls.readers, key) + ls.mu.Unlock() + cancel() + log.Debug().Err(err).Str("container", c.ID).Str("host", c.Host).Msg("log streamer: could not find container, skipping") + return + } + + ls.wg.Add(1) + go func() { + defer ls.wg.Done() + defer func() { + ls.mu.Lock() + delete(ls.readers, key) + ls.mu.Unlock() + cancel() + }() + ls.runReader(readerCtx, cs) + }() +} + +// runReader follows logs from a single container and pushes batches directly +// to the cloud via ls.send. send() is serialised by the caller, so a slow +// cloud connection backpressures all readers — this is intentional. +func (ls *logStreamer) runReader(ctx context.Context, cs *container_support.ContainerService) { + events := make(chan *container.LogEvent, logReaderChanBuffer) + + streamErr := make(chan error, 1) + go func() { + defer close(events) + // Start from "now" to avoid replaying historical logs on every reconnect. + streamErr <- cs.StreamLogs(ctx, time.Now(), container.STDOUT|container.STDERR, events) + }() + + hostID := cs.Container.Host + containerID := cs.Container.ID + containerName := cs.Container.Name + + log.Debug().Str("container", containerName).Str("host", hostID).Msg("log streamer: reader started") + + var batch []*pb.LogBatchEntry + var batchBytes int + flushTicker := time.NewTicker(logBatchFlushPeriod) + defer flushTicker.Stop() + + flush := func() error { + if len(batch) == 0 { + return nil + } + err := ls.send(&pb.ToolResponse{Type: &pb.ToolResponse_LogBatch{LogBatch: &pb.LogBatch{Entries: batch}}}) + batch = nil + batchBytes = 0 + return err + } + + defer func() { + _ = flush() + log.Debug().Str("container", containerName).Str("host", hostID).Msg("log streamer: reader stopped") + }() + + for { + select { + case <-ctx.Done(): + return + case <-flushTicker.C: + if err := flush(); err != nil { + log.Debug().Err(err).Msg("log streamer: send failed") + return + } + case ev, ok := <-events: + if !ok { + if err := <-streamErr; err != nil && ctx.Err() == nil { + log.Debug().Err(err).Str("container", containerName).Msg("log streamer: StreamLogs ended with error") + } + return + } + + msg := ev.RawMessage + if msg == "" { + msg = messageToString(ev.Message) + } + + tsNs := ev.Timestamp * int64(time.Millisecond) // LogEvent.Timestamp is UnixMilli + if tsNs == 0 { + tsNs = time.Now().UnixNano() + } + + level := ev.Level + if level == "unknown" { + level = "" + } + + batch = append(batch, &pb.LogBatchEntry{ + HostId: hostID, + ContainerId: containerID, + ContainerName: containerName, + TimestampNs: tsNs, + Message: msg, + Stream: ev.Stream, + Level: level, + }) + batchBytes += len(msg) + + if len(batch) >= logBatchMaxEntries || batchBytes >= logBatchMaxBytes { + if err := flush(); err != nil { + log.Debug().Err(err).Msg("log streamer: send failed") + return + } + } + } + } +} + +// messageToString renders a LogEvent.Message of any concrete type into a +// string suitable for transport. Grouped multi-line events don't set +// RawMessage, so JSON-encode their fragment slice as a fallback. +func messageToString(m any) string { + switch v := m.(type) { + case nil: + return "" + case string: + return v + default: + b, err := json.Marshal(v) + if err != nil { + log.Debug().Err(err).Msg("log streamer: failed to marshal message") + return "" + } + return string(b) + } +} diff --git a/internal/cloud/log_streamer_test.go b/internal/cloud/log_streamer_test.go new file mode 100644 index 00000000..81fe609c --- /dev/null +++ b/internal/cloud/log_streamer_test.go @@ -0,0 +1,358 @@ +package cloud + +import ( + "context" + "io" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/amir20/dozzle/internal/container" + container_support "github.com/amir20/dozzle/internal/support/container" + pb "github.com/amir20/dozzle/proto/cloud" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeClientService is a ClientService that delivers scripted log events. +type fakeClientService struct { + host container.Host + logsCh chan *container.LogEvent + streamed atomic.Bool + wait chan struct{} // closed once StreamLogs is invoked + once sync.Once +} + +func newFakeClientService(hostID string) *fakeClientService { + return &fakeClientService{ + host: container.Host{ID: hostID, Name: hostID}, + logsCh: make(chan *container.LogEvent, 16), + wait: make(chan struct{}), + } +} + +func (f *fakeClientService) FindContainer(_ context.Context, _ string, _ container.ContainerLabels) (container.Container, error) { + return container.Container{}, nil +} +func (f *fakeClientService) ListContainers(_ context.Context, _ container.ContainerLabels) ([]container.Container, error) { + return nil, nil +} +func (f *fakeClientService) Host(_ context.Context) (container.Host, error) { return f.host, nil } +func (f *fakeClientService) ContainerAction(_ context.Context, _ container.Container, _ container.ContainerAction) error { + return nil +} +func (f *fakeClientService) UpdateContainer(_ context.Context, _ container.Container, progressCh chan<- container.UpdateProgress) (bool, error) { + close(progressCh) + return false, nil +} +func (f *fakeClientService) LogsBetweenDates(_ context.Context, _ container.Container, _ time.Time, _ time.Time, _ container.StdType) (<-chan *container.LogEvent, error) { + return nil, nil +} +func (f *fakeClientService) RawLogs(_ context.Context, _ container.Container, _ time.Time, _ time.Time, _ container.StdType) (io.ReadCloser, error) { + return nil, nil +} +func (f *fakeClientService) SubscribeStats(_ context.Context, _ chan<- container.ContainerStat) {} +func (f *fakeClientService) SubscribeEvents(_ context.Context, _ chan<- container.ContainerEvent) {} +func (f *fakeClientService) SubscribeContainersStarted(_ context.Context, _ chan<- container.Container) { +} +func (f *fakeClientService) StreamLogs(ctx context.Context, _ container.Container, _ time.Time, _ container.StdType, events chan<- *container.LogEvent) error { + f.once.Do(func() { close(f.wait) }) + f.streamed.Store(true) + for { + select { + case <-ctx.Done(): + return ctx.Err() + case ev, ok := <-f.logsCh: + if !ok { + return io.EOF + } + select { + case events <- ev: + case <-ctx.Done(): + return ctx.Err() + } + } + } +} +func (f *fakeClientService) Attach(_ context.Context, _ container.Container, _ container.ExecEventReader, _ io.Writer) error { + return nil +} +func (f *fakeClientService) Exec(_ context.Context, _ container.Container, _ []string, _ container.ExecEventReader, _ io.Writer) error { + return nil +} + +// fakeHostService is a LogStreamHostService driven from a map of containers. +type fakeHostService struct { + mu sync.Mutex + containers []container.Container + clients map[string]*fakeClientService // hostID -> client + startedCh chan<- container.Container +} + +func (f *fakeHostService) ListAllContainers(_ container.ContainerLabels) ([]container.Container, []error) { + f.mu.Lock() + defer f.mu.Unlock() + return append([]container.Container(nil), f.containers...), nil +} + +func (f *fakeHostService) FindContainer(host string, id string, _ container.ContainerLabels) (*container_support.ContainerService, error) { + f.mu.Lock() + defer f.mu.Unlock() + client, ok := f.clients[host] + if !ok { + return nil, assert.AnError + } + for _, c := range f.containers { + if c.ID == id && c.Host == host { + return container_support.NewContainerService(client, c), nil + } + } + return nil, assert.AnError +} + +func (f *fakeHostService) Hosts() []container.Host { + f.mu.Lock() + defer f.mu.Unlock() + hosts := make([]container.Host, 0, len(f.clients)) + for id := range f.clients { + hosts = append(hosts, container.Host{ID: id, Name: id}) + } + return hosts +} + +func (f *fakeHostService) SubscribeContainersStarted(_ context.Context, ch chan<- container.Container, _ container_support.ContainerFilter) { + f.mu.Lock() + f.startedCh = ch + f.mu.Unlock() +} + +func (f *fakeHostService) emitStart(c container.Container) { + f.mu.Lock() + f.containers = append(f.containers, c) + ch := f.startedCh + f.mu.Unlock() + if ch != nil { + ch <- c + } +} + +func collectBatches(t *testing.T, mu *sync.Mutex, out *[]*pb.LogBatch, wantEntries int, timeout time.Duration) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + mu.Lock() + total := 0 + for _, b := range *out { + total += len(b.Entries) + } + mu.Unlock() + if total >= wantEntries { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("timed out waiting for %d entries", wantEntries) +} + +func TestLogStreamer_InitialSnapshotAndBatching(t *testing.T) { + client := newFakeClientService("host-1") + hs := &fakeHostService{ + containers: []container.Container{ + {ID: "c1", Name: "nginx", Host: "host-1", State: "running"}, + }, + clients: map[string]*fakeClientService{"host-1": client}, + } + + var sendMu sync.Mutex + var sent []*pb.LogBatch + send := func(resp *pb.ToolResponse) error { + sendMu.Lock() + defer sendMu.Unlock() + if lb := resp.GetLogBatch(); lb != nil { + sent = append(sent, lb) + } + return nil + } + + ls := newLogStreamer(hs, nil, send) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + runDone := make(chan struct{}) + go func() { + ls.run(ctx) + close(runDone) + }() + + // Wait for reader to hook in. + select { + case <-client.wait: + case <-time.After(2 * time.Second): + t.Fatal("StreamLogs was never called") + } + + ts := time.Now().UnixMilli() + for i := 0; i < 3; i++ { + client.logsCh <- &container.LogEvent{ + Timestamp: ts, + RawMessage: "hello world", + Stream: "stdout", + Level: "info", + } + } + + collectBatches(t, &sendMu, &sent, 3, 3*time.Second) + + sendMu.Lock() + var allEntries []*pb.LogBatchEntry + for _, b := range sent { + allEntries = append(allEntries, b.Entries...) + } + sendMu.Unlock() + + require.GreaterOrEqual(t, len(allEntries), 3) + e := allEntries[0] + assert.Equal(t, "host-1", e.HostId) + assert.Equal(t, "c1", e.ContainerId) + assert.Equal(t, "nginx", e.ContainerName) + assert.Equal(t, "hello world", e.Message) + assert.Equal(t, "stdout", e.Stream) + assert.Equal(t, "info", e.Level) + assert.Equal(t, ts*int64(time.Millisecond), e.TimestampNs) + + cancel() + <-runDone +} + +func TestLogStreamer_NewContainerStartsReader(t *testing.T) { + client := newFakeClientService("host-1") + hs := &fakeHostService{ + containers: nil, + clients: map[string]*fakeClientService{"host-1": client}, + } + + send := func(_ *pb.ToolResponse) error { return nil } + ls := newLogStreamer(hs, nil, send) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + runDone := make(chan struct{}) + go func() { + ls.run(ctx) + close(runDone) + }() + + // Wait for run() to subscribe before emitting; emitting before + // startedCh is registered would silently no-op. + require.Eventually(t, func() bool { + hs.mu.Lock() + defer hs.mu.Unlock() + return hs.startedCh != nil + }, 2*time.Second, 5*time.Millisecond, "subscription was never registered") + + hs.emitStart(container.Container{ID: "c-new", Name: "redis", Host: "host-1", State: "running"}) + + select { + case <-client.wait: + case <-time.After(2 * time.Second): + t.Fatal("reader was not started for new container") + } + + cancel() + <-runDone +} + +func TestLogStreamer_LevelUnknownIsBlank(t *testing.T) { + client := newFakeClientService("host-1") + hs := &fakeHostService{ + containers: []container.Container{ + {ID: "c1", Name: "n", Host: "host-1", State: "running"}, + }, + clients: map[string]*fakeClientService{"host-1": client}, + } + + var sendMu sync.Mutex + var sent []*pb.LogBatch + send := func(resp *pb.ToolResponse) error { + sendMu.Lock() + defer sendMu.Unlock() + if lb := resp.GetLogBatch(); lb != nil { + sent = append(sent, lb) + } + return nil + } + ls := newLogStreamer(hs, nil, send) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go ls.run(ctx) + + <-client.wait + client.logsCh <- &container.LogEvent{Timestamp: time.Now().UnixMilli(), RawMessage: "m", Stream: "stdout", Level: "unknown"} + + collectBatches(t, &sendMu, &sent, 1, 2*time.Second) + sendMu.Lock() + assert.Equal(t, "", sent[0].Entries[0].Level) + sendMu.Unlock() +} + +func TestLogStreamer_BatchFlushesOnMaxEntries(t *testing.T) { + client := newFakeClientService("host-1") + hs := &fakeHostService{ + containers: []container.Container{ + {ID: "c1", Name: "n", Host: "host-1", State: "running"}, + }, + clients: map[string]*fakeClientService{"host-1": client}, + } + + var sendMu sync.Mutex + var sent []*pb.LogBatch + send := func(resp *pb.ToolResponse) error { + sendMu.Lock() + defer sendMu.Unlock() + if lb := resp.GetLogBatch(); lb != nil { + sent = append(sent, lb) + } + return nil + } + ls := newLogStreamer(hs, nil, send) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + runDone := make(chan struct{}) + go func() { ls.run(ctx); close(runDone) }() + + <-client.wait + // Push just over the max entries cap so we flush before the 1s timer. + ts := time.Now().UnixMilli() + total := logBatchMaxEntries + 10 + for i := 0; i < total; i++ { + client.logsCh <- &container.LogEvent{Timestamp: ts, RawMessage: "x", Stream: "stdout", Level: "info"} + } + + // A full flush should happen well under 1s (we're below the timer). + deadline := time.Now().Add(500 * time.Millisecond) + for time.Now().Before(deadline) { + sendMu.Lock() + hasFullBatch := false + for _, b := range sent { + if len(b.Entries) >= logBatchMaxEntries { + hasFullBatch = true + break + } + } + sendMu.Unlock() + if hasFullBatch { + cancel() + <-runDone + return + } + time.Sleep(10 * time.Millisecond) + } + cancel() + <-runDone + t.Fatal("expected a batch with >= logBatchMaxEntries entries") +} diff --git a/internal/notification/cloud_config.go b/internal/notification/cloud_config.go index e3e99752..ac42fe17 100644 --- a/internal/notification/cloud_config.go +++ b/internal/notification/cloud_config.go @@ -12,6 +12,19 @@ type CloudConfig struct { APIKey string `yaml:"apiKey"` Prefix string `yaml:"prefix"` ExpiresAt *time.Time `yaml:"expiresAt,omitempty"` + // StreamLogs controls whether container logs are streamed to Dozzle Cloud. + // nil means default (enabled) — preserves behavior for configs written + // before this field existed. + StreamLogs *bool `yaml:"streamLogs,omitempty"` +} + +// StreamLogsEnabled reports whether the bulk log stream to cloud should run. +// Defaults to true when the field is unset or the config is nil. +func (c *CloudConfig) StreamLogsEnabled() bool { + if c == nil || c.StreamLogs == nil { + return true + } + return *c.StreamLogs } // WriteCloudConfig encodes the given CloudConfig to the writer in YAML format. diff --git a/internal/notification/persist.go b/internal/notification/persist.go index 566b0482..6e323270 100644 --- a/internal/notification/persist.go +++ b/internal/notification/persist.go @@ -113,6 +113,20 @@ func (p *Persister) SetCloudConfig(cc *CloudConfig) { p.SaveCloud() } +// SetCloudStreamLogs updates the StreamLogs flag on the current cloud config +// and persists it. No-op when no cloud config is set. +func (p *Persister) SetCloudStreamLogs(enabled bool) { + p.mu.Lock() + if p.cloudConfig == nil { + p.mu.Unlock() + return + } + v := enabled + p.cloudConfig.StreamLogs = &v + p.mu.Unlock() + p.SaveCloud() +} + // RemoveCloudConfig clears the cloud config, clears the cloud dispatcher, and // removes the cloud config file from disk. func (p *Persister) RemoveCloudConfig() { diff --git a/internal/support/docker/multi_host_service.go b/internal/support/docker/multi_host_service.go index afbbfbb1..80fdc1c7 100644 --- a/internal/support/docker/multi_host_service.go +++ b/internal/support/docker/multi_host_service.go @@ -248,6 +248,13 @@ func (m *MultiHostService) SetCloudConfig(cc *notification.CloudConfig) { m.broadcastCloudConfig() } +// SetCloudStreamLogs updates the bulk-log-streaming privacy flag on the cloud +// config and persists it. Only affects the local cloud client; agents never +// stream logs directly to cloud. +func (m *MultiHostService) SetCloudStreamLogs(enabled bool) { + m.persister.SetCloudStreamLogs(enabled) +} + // RemoveCloudConfig clears the cloud config, removes the cloud dispatcher, deletes the file, // and broadcasts the change to all agents so they stop sending to cloud. func (m *MultiHostService) RemoveCloudConfig() { diff --git a/internal/support/k8s/k8s_cluster_service.go b/internal/support/k8s/k8s_cluster_service.go index e4af7b2c..fffcec18 100644 --- a/internal/support/k8s/k8s_cluster_service.go +++ b/internal/support/k8s/k8s_cluster_service.go @@ -231,6 +231,10 @@ func (m *K8sClusterService) SetCloudConfig(cc *notification.CloudConfig) { m.persister.SetCloudConfig(cc) } +func (m *K8sClusterService) SetCloudStreamLogs(enabled bool) { + m.persister.SetCloudStreamLogs(enabled) +} + func (m *K8sClusterService) RemoveCloudConfig() { m.persister.RemoveCloudConfig() } diff --git a/internal/web/cloud.go b/internal/web/cloud.go index 03ccf999..02c6a785 100644 --- a/internal/web/cloud.go +++ b/internal/web/cloud.go @@ -180,9 +180,10 @@ func (h *handler) cloudStatus(w http.ResponseWriter, r *http.Request) { } type cloudConfigResponse struct { - Prefix string `json:"prefix"` - ExpiresAt *string `json:"expiresAt,omitempty"` - Linked bool `json:"linked"` + Prefix string `json:"prefix"` + ExpiresAt *string `json:"expiresAt,omitempty"` + Linked bool `json:"linked"` + StreamLogs bool `json:"streamLogs"` } func (h *handler) cloudConfig(w http.ResponseWriter, r *http.Request) { @@ -193,8 +194,9 @@ func (h *handler) cloudConfig(w http.ResponseWriter, r *http.Request) { } resp := cloudConfigResponse{ - Prefix: cc.Prefix, - Linked: true, + Prefix: cc.Prefix, + Linked: true, + StreamLogs: cc.StreamLogsEnabled(), } if cc.ExpiresAt != nil { s := cc.ExpiresAt.Format(time.RFC3339) @@ -206,6 +208,35 @@ func (h *handler) cloudConfig(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(resp) } +type updateCloudConfigRequest struct { + StreamLogs *bool `json:"streamLogs,omitempty"` +} + +func (h *handler) updateCloudConfig(w http.ResponseWriter, r *http.Request) { + cc := h.hostService.CloudConfig() + if cc == nil { + writeError(w, http.StatusNotFound, "no cloud configuration") + return + } + + var req updateCloudConfigRequest + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<10)).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid request body") + return + } + + if req.StreamLogs != nil { + h.hostService.SetCloudStreamLogs(*req.StreamLogs) + // Drop the active cloud connection so the new flag is picked up on + // the next dial — a streamer may need to start or stop. + if h.config.OnCloudUpdate != nil { + h.config.OnCloudUpdate() + } + } + + w.WriteHeader(http.StatusNoContent) +} + func (h *handler) deleteCloudConfig(w http.ResponseWriter, r *http.Request) { h.hostService.RemoveCloudConfig() w.WriteHeader(http.StatusNoContent) diff --git a/internal/web/routes.go b/internal/web/routes.go index e9af1de6..6dbd5fa9 100644 --- a/internal/web/routes.go +++ b/internal/web/routes.go @@ -51,6 +51,7 @@ type Config struct { ReleaseCheckMode ReleaseCheckMode Labels container.ContainerLabels OnCloudSetup func() + OnCloudUpdate func() } type Authorization struct { @@ -90,6 +91,7 @@ type HostService interface { FetchAgentNotificationStats() map[int]types.SubscriptionStats CloudConfig() *notification.CloudConfig SetCloudConfig(cc *notification.CloudConfig) + SetCloudStreamLogs(enabled bool) RemoveCloudConfig() } @@ -190,6 +192,7 @@ func createRouter(h *handler) *chi.Mux { // Cloud API r.Get("/cloud/status", h.cloudStatus) r.Get("/cloud/config", h.cloudConfig) + r.Patch("/cloud/config", h.updateCloudConfig) r.Delete("/cloud/config", h.deleteCloudConfig) r.Post("/cloud/feedback", h.cloudFeedback) }) diff --git a/locales/da.yml b/locales/da.yml index 5cab46de..bb255e26 100644 --- a/locales/da.yml +++ b/locales/da.yml @@ -304,6 +304,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: Fjern tilknytning unlink-confirm: Er du sikker på, at du vil fjerne tilknytningen til Dozzle Cloud? Dette fjerner alle cloud-notifikationsdestinationer. + stream-logs: Stream containerlogs til Dozzle Cloud + stream-logs-help: Påkrævet for AI-drevne undersøgelser og logsøgning. Deaktiver for at holde alt logindhold på denne instans. welcome: title: "Din instans er forbundet!" subtitle: "Du er klar til at modtage advarsler, daglige resuméer og mere fra Dozzle Cloud." diff --git a/locales/de.yml b/locales/de.yml index a81a116c..465228ad 100644 --- a/locales/de.yml +++ b/locales/de.yml @@ -304,6 +304,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: Verknüpfung aufheben unlink-confirm: Sind Sie sicher, dass Sie die Verknüpfung mit Dozzle Cloud aufheben möchten? Dies entfernt alle Cloud-Benachrichtigungsziele. + stream-logs: Container-Logs an Dozzle Cloud streamen + stream-logs-help: Erforderlich für KI-gestützte Analysen und Log-Suche. Deaktivieren, um sämtliche Log-Inhalte auf dieser Instanz zu behalten. welcome: title: "Ihre Instanz ist verbunden!" subtitle: "Sie können jetzt Benachrichtigungen, tägliche Zusammenfassungen und mehr von Dozzle Cloud erhalten." diff --git a/locales/en.yml b/locales/en.yml index 8ae96478..1583c717 100644 --- a/locales/en.yml +++ b/locales/en.yml @@ -317,6 +317,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: Unlink unlink-confirm: Are you sure you want to unlink from Dozzle Cloud? This will remove all cloud notification destinations. + stream-logs: Stream container logs to Dozzle Cloud + stream-logs-help: Required for AI-powered investigations and log search. Disable to keep all log content on this instance. welcome: title: "Your instance is connected!" subtitle: "You're all set to start getting alerts, daily digests, and more from Dozzle Cloud." diff --git a/locales/es.yml b/locales/es.yml index e1cbb811..9333b215 100644 --- a/locales/es.yml +++ b/locales/es.yml @@ -304,6 +304,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: Desvincular unlink-confirm: ¿Está seguro de que desea desvincular de Dozzle Cloud? Esto eliminará todos los destinos de notificación en la nube. + stream-logs: Transmitir registros de contenedores a Dozzle Cloud + stream-logs-help: Necesario para investigaciones con IA y búsqueda de registros. Desactívelo para mantener todo el contenido de los registros en esta instancia. welcome: title: "¡Tu instancia está conectada!" subtitle: "Ya puedes recibir alertas, resúmenes diarios y más desde Dozzle Cloud." diff --git a/locales/fr.yml b/locales/fr.yml index c7445bbd..04b80498 100644 --- a/locales/fr.yml +++ b/locales/fr.yml @@ -304,6 +304,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: Délier unlink-confirm: Êtes-vous sûr de vouloir délier de Dozzle Cloud ? Cela supprimera toutes les destinations de notification cloud. + stream-logs: Diffuser les journaux de conteneurs vers Dozzle Cloud + stream-logs-help: Requis pour les investigations alimentées par IA et la recherche dans les journaux. Désactivez pour conserver tout le contenu des journaux sur cette instance. welcome: title: "Votre instance est connectée !" subtitle: "Vous êtes prêt à recevoir des alertes, des résumés quotidiens et bien plus depuis Dozzle Cloud." diff --git a/locales/id.yml b/locales/id.yml index efc8e9f8..d18c6668 100644 --- a/locales/id.yml +++ b/locales/id.yml @@ -316,6 +316,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: Lepaskan tautan unlink-confirm: Apakah Anda yakin ingin melepaskan tautan dari Dozzle Cloud? Ini akan menghapus semua tujuan notifikasi cloud. + stream-logs: Streaming log kontainer ke Dozzle Cloud + stream-logs-help: Diperlukan untuk investigasi bertenaga AI dan pencarian log. Nonaktifkan untuk menyimpan seluruh konten log di instance ini. welcome: title: "Instans Anda terhubung!" subtitle: "Anda siap menerima peringatan, ringkasan harian, dan lainnya dari Dozzle Cloud." diff --git a/locales/it.yml b/locales/it.yml index f90b7631..3e070642 100644 --- a/locales/it.yml +++ b/locales/it.yml @@ -304,6 +304,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: Scollega unlink-confirm: Sei sicuro di voler scollegare da Dozzle Cloud? Questo rimuoverà tutte le destinazioni di notifica cloud. + stream-logs: Invia i log dei container a Dozzle Cloud + stream-logs-help: Necessario per le indagini basate sull'IA e la ricerca nei log. Disattiva per mantenere tutto il contenuto dei log su questa istanza. welcome: title: "La tua istanza è connessa!" subtitle: "Sei pronto per ricevere avvisi, riepiloghi giornalieri e altro da Dozzle Cloud." diff --git a/locales/ko.yml b/locales/ko.yml index 81d7c98c..12507346 100644 --- a/locales/ko.yml +++ b/locales/ko.yml @@ -307,6 +307,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: 연결 해제 unlink-confirm: Dozzle Cloud 연결을 해제하시겠습니까? 모든 클라우드 알림 목적지가 삭제됩니다. + stream-logs: 컨테이너 로그를 Dozzle Cloud로 전송 + stream-logs-help: AI 기반 분석 및 로그 검색에 필요합니다. 모든 로그 내용을 이 인스턴스에 유지하려면 비활성화하세요. welcome: title: "인스턴스가 연결되었습니다!" subtitle: "이제 Dozzle Cloud에서 알림, 일일 요약 등을 받을 수 있습니다." diff --git a/locales/nl.yml b/locales/nl.yml index 9f228906..0e8c12f4 100644 --- a/locales/nl.yml +++ b/locales/nl.yml @@ -305,6 +305,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: Ontkoppelen unlink-confirm: Weet je zeker dat je de koppeling met Dozzle Cloud wilt opheffen? Dit verwijdert alle cloudmeldingsbestemmingen. + stream-logs: Containerlogs streamen naar Dozzle Cloud + stream-logs-help: Vereist voor AI-gestuurd onderzoek en zoeken in logs. Schakel uit om alle loginhoud op deze instantie te houden. welcome: title: "Je instantie is verbonden!" subtitle: "Je kunt nu meldingen, dagelijkse samenvattingen en meer ontvangen van Dozzle Cloud." diff --git a/locales/pl.yml b/locales/pl.yml index 17c05f49..bf7d09e3 100644 --- a/locales/pl.yml +++ b/locales/pl.yml @@ -311,6 +311,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: Rozłącz unlink-confirm: Czy na pewno chcesz rozłączyć się z Dozzle Cloud? Spowoduje to usunięcie wszystkich miejsc docelowych powiadomień w chmurze. + stream-logs: Przesyłaj logi kontenerów do Dozzle Cloud + stream-logs-help: Wymagane do analiz wspieranych przez AI i wyszukiwania w logach. Wyłącz, aby zachować całą zawartość logów w tej instancji. welcome: title: "Twoja instancja jest połączona!" subtitle: "Możesz teraz otrzymywać alerty, codzienne podsumowania i więcej z Dozzle Cloud." diff --git a/locales/pr.yml b/locales/pr.yml index 5f67b299..fe6053b2 100644 --- a/locales/pr.yml +++ b/locales/pr.yml @@ -313,6 +313,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: Desligar unlink-confirm: Tem a certeza de que pretende desligar do Dozzle Cloud? Isto irá remover todos os destinos de notificação na nuvem. + stream-logs: Transmitir registos de contentores para a Dozzle Cloud + stream-logs-help: Necessário para investigações com IA e pesquisa de registos. Desative para manter todo o conteúdo dos registos nesta instância. welcome: title: "Ahoy! Yer ship be connected!" subtitle: "Ye be ready to receive alerts, daily summaries, and more from Dozzle Cloud, ye scallywag." diff --git a/locales/pt.yml b/locales/pt.yml index 11eef805..e0aa0ff2 100644 --- a/locales/pt.yml +++ b/locales/pt.yml @@ -303,6 +303,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: Desvincular unlink-confirm: Tem certeza de que deseja desvincular do Dozzle Cloud? Isso removerá todos os destinos de notificação na nuvem. + stream-logs: Transmitir logs de contêineres para o Dozzle Cloud + stream-logs-help: Necessário para investigações com IA e busca em logs. Desative para manter todo o conteúdo dos logs nesta instância. welcome: title: "Sua instância está conectada!" subtitle: "Você já pode receber alertas, resumos diários e mais do Dozzle Cloud." diff --git a/locales/ru.yml b/locales/ru.yml index 3ad6fee7..d607ce4e 100644 --- a/locales/ru.yml +++ b/locales/ru.yml @@ -304,6 +304,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: Отвязать unlink-confirm: Вы уверены, что хотите отвязать от Dozzle Cloud? Это удалит все облачные назначения уведомлений. + stream-logs: Передавать логи контейнеров в Dozzle Cloud + stream-logs-help: Требуется для расследований с помощью ИИ и поиска по логам. Отключите, чтобы все содержимое логов оставалось на этом экземпляре. welcome: title: "Ваш экземпляр подключён!" subtitle: "Вы готовы получать оповещения, ежедневные сводки и многое другое от Dozzle Cloud." diff --git a/locales/sl.yml b/locales/sl.yml index 163d0d10..c48a3bc1 100644 --- a/locales/sl.yml +++ b/locales/sl.yml @@ -309,6 +309,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: Prekini povezavo unlink-confirm: Ali ste prepričani, da želite prekiniti povezavo z Dozzle Cloud? To bo odstranilo vse cilje obvestil v oblaku. + stream-logs: Pretakanje dnevnikov vsebnikov v Dozzle Cloud + stream-logs-help: Potrebno za preiskave z umetno inteligenco in iskanje po dnevnikih. Onemogočite, da vsa vsebina dnevnikov ostane na tej instanci. welcome: title: "Vaša instanca je povezana!" subtitle: "Zdaj lahko prejemate opozorila, dnevne povzetke in več iz Dozzle Cloud." diff --git a/locales/tr.yml b/locales/tr.yml index d9738de7..76764519 100644 --- a/locales/tr.yml +++ b/locales/tr.yml @@ -304,6 +304,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: Bağlantıyı kaldır unlink-confirm: Dozzle Cloud bağlantısını kaldırmak istediğinizden emin misiniz? Bu, tüm bulut bildirim hedeflerini kaldıracaktır. + stream-logs: Konteyner günlüklerini Dozzle Cloud'a yayınla + stream-logs-help: Yapay zeka destekli incelemeler ve günlük araması için gereklidir. Tüm günlük içeriğini bu örnekte tutmak için devre dışı bırakın. welcome: title: "Örneğiniz bağlandı!" subtitle: "Artık Dozzle Cloud'dan uyarılar, günlük özetler ve daha fazlasını alabilirsiniz." diff --git a/locales/zh-tw.yml b/locales/zh-tw.yml index c338160e..86bb5347 100644 --- a/locales/zh-tw.yml +++ b/locales/zh-tw.yml @@ -307,6 +307,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: 取消連結 unlink-confirm: 確定要取消與 Dozzle Cloud 的連結嗎？這將移除所有雲端通知目標。 + stream-logs: 將容器日誌串流至 Dozzle Cloud + stream-logs-help: AI 分析與日誌搜尋所需。停用後，所有日誌內容將保留在此執行個體上。 welcome: title: "您的實例已連線！" subtitle: "您現在可以從 Dozzle Cloud 接收警示、每日摘要等。" diff --git a/locales/zh.yml b/locales/zh.yml index 694baa61..2e12259d 100644 --- a/locales/zh.yml +++ b/locales/zh.yml @@ -304,6 +304,8 @@ cloud: error-unavailable: Dozzle Cloud is temporarily unavailable. Please try again later. unlink: 取消关联 unlink-confirm: 确定要取消与 Dozzle Cloud 的关联吗？这将移除所有云通知目标。 + stream-logs: 将容器日志流式传输到 Dozzle Cloud + stream-logs-help: AI 调查和日志搜索所需。禁用后，所有日志内容将保留在此实例中。 welcome: title: "您的实例已连接！" subtitle: "您现在可以从 Dozzle Cloud 接收警报、每日摘要等。" diff --git a/main.go b/main.go index 00b87d46..33e1a644 100644 --- a/main.go +++ b/main.go @@ -160,6 +160,9 @@ func main() { DeployManager: deployManager, NotificationService: notificationService, }) + cloudClient.SetStreamLogsFunc(func() bool { + return hostService.CloudConfig().StreamLogsEnabled() + }) go cloudClient.Run(ctx) // If cloud is already configured at startup, start the client immediately @@ -167,7 +170,7 @@ func main() { cloudClient.Notify() } - srv := createServer(args, hostService, cloudClient.Notify) + srv := createServer(args, hostService, cloudClient.Notify, cloudClient.Reconnect) go func() { log.Info().Msgf("Accepting connections on %s", args.Addr) @@ -195,7 +198,7 @@ func fileExists(filename string) bool { return err == nil } -func createServer(args cli.Args, hostService web.HostService, onCloudSetup func()) *http.Server { +func createServer(args cli.Args, hostService web.HostService, onCloudSetup func(), onCloudUpdate func()) *http.Server { _, dev := os.LookupEnv("DEV") var releaseCheckMode web.ReleaseCheckMode = web.Automatic @@ -275,6 +278,7 @@ func createServer(args cli.Args, hostService web.HostService, onCloudSetup func( ReleaseCheckMode: releaseCheckMode, Labels: args.Filter, OnCloudSetup: onCloudSetup, + OnCloudUpdate: onCloudUpdate, } assets, err := fs.Sub(content, "dist") diff --git a/proto/cloud/cloud.pb.go b/proto/cloud/cloud.pb.go index 91f989e8..ee286f45 100644 --- a/proto/cloud/cloud.pb.go +++ b/proto/cloud/cloud.pb.go @@ -195,6 +195,7 @@ type ToolResponse struct { // // *ToolResponse_ListTools // *ToolResponse_CallTool + // *ToolResponse_LogBatch Type isToolResponse_Type `protobuf_oneof:"type"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -262,6 +263,15 @@ func (x *ToolResponse) GetCallTool() *CallToolResponse { return nil } +func (x *ToolResponse) GetLogBatch() *LogBatch { + if x != nil { + if x, ok := x.Type.(*ToolResponse_LogBatch); ok { + return x.LogBatch + } + } + return nil +} + type isToolResponse_Type interface { isToolResponse_Type() } @@ -274,10 +284,158 @@ type ToolResponse_CallTool struct { CallTool *CallToolResponse `protobuf:"bytes,3,opt,name=call_tool,json=callTool,proto3,oneof"` } +type ToolResponse_LogBatch struct { + // Unsolicited server-push: batched container log lines streamed from + // Dozzle to Cloud for ingestion into VictoriaLogs. request_id is empty + // for log batches — they are not replies to a ToolRequest. + LogBatch *LogBatch `protobuf:"bytes,4,opt,name=log_batch,json=logBatch,proto3,oneof"` +} + func (*ToolResponse_ListTools) isToolResponse_Type() {} func (*ToolResponse_CallTool) isToolResponse_Type() {} +func (*ToolResponse_LogBatch) isToolResponse_Type() {} + +// Batch of log lines from one or more containers. Dozzle pushes these +// continuously while connected. Cloud routes them to VictoriaLogs scoped +// to the owning user (derived from the connection's auth). +type LogBatch struct { + state protoimpl.MessageState `protogen:"open.v1"` + Entries []*LogBatchEntry `protobuf:"bytes,1,rep,name=entries,proto3" json:"entries,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *LogBatch) Reset() { + *x = LogBatch{} + mi := &file_cloud_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *LogBatch) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*LogBatch) ProtoMessage() {} + +func (x *LogBatch) ProtoReflect() protoreflect.Message { + mi := &file_cloud_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use LogBatch.ProtoReflect.Descriptor instead. +func (*LogBatch) Descriptor() ([]byte, []int) { + return file_cloud_proto_rawDescGZIP(), []int{2} +} + +func (x *LogBatch) GetEntries() []*LogBatchEntry { + if x != nil { + return x.Entries + } + return nil +} + +type LogBatchEntry struct { + state protoimpl.MessageState `protogen:"open.v1"` + HostId string `protobuf:"bytes,1,opt,name=host_id,json=hostId,proto3" json:"host_id,omitempty"` + ContainerId string `protobuf:"bytes,2,opt,name=container_id,json=containerId,proto3" json:"container_id,omitempty"` + ContainerName string `protobuf:"bytes,3,opt,name=container_name,json=containerName,proto3" json:"container_name,omitempty"` + TimestampNs int64 `protobuf:"varint,4,opt,name=timestamp_ns,json=timestampNs,proto3" json:"timestamp_ns,omitempty"` // unix nanoseconds + Message string `protobuf:"bytes,5,opt,name=message,proto3" json:"message,omitempty"` + Stream string `protobuf:"bytes,6,opt,name=stream,proto3" json:"stream,omitempty"` // "stdout" or "stderr" + Level string `protobuf:"bytes,7,opt,name=level,proto3" json:"level,omitempty"` // "info", "warn", "error", etc. (best-effort) + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *LogBatchEntry) Reset() { + *x = LogBatchEntry{} + mi := &file_cloud_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *LogBatchEntry) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*LogBatchEntry) ProtoMessage() {} + +func (x *LogBatchEntry) ProtoReflect() protoreflect.Message { + mi := &file_cloud_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use LogBatchEntry.ProtoReflect.Descriptor instead. +func (*LogBatchEntry) Descriptor() ([]byte, []int) { + return file_cloud_proto_rawDescGZIP(), []int{3} +} + +func (x *LogBatchEntry) GetHostId() string { + if x != nil { + return x.HostId + } + return "" +} + +func (x *LogBatchEntry) GetContainerId() string { + if x != nil { + return x.ContainerId + } + return "" +} + +func (x *LogBatchEntry) GetContainerName() string { + if x != nil { + return x.ContainerName + } + return "" +} + +func (x *LogBatchEntry) GetTimestampNs() int64 { + if x != nil { + return x.TimestampNs + } + return 0 +} + +func (x *LogBatchEntry) GetMessage() string { + if x != nil { + return x.Message + } + return "" +} + +func (x *LogBatchEntry) GetStream() string { + if x != nil { + return x.Stream + } + return "" +} + +func (x *LogBatchEntry) GetLevel() string { + if x != nil { + return x.Level + } + return "" +} + type ListToolsRequest struct { state protoimpl.MessageState `protogen:"open.v1"` unknownFields protoimpl.UnknownFields @@ -286,7 +444,7 @@ type ListToolsRequest struct { func (x *ListToolsRequest) Reset() { *x = ListToolsRequest{} - mi := &file_cloud_proto_msgTypes[2] + mi := &file_cloud_proto_msgTypes[4] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -298,7 +456,7 @@ func (x *ListToolsRequest) String() string { func (*ListToolsRequest) ProtoMessage() {} func (x *ListToolsRequest) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[2] + mi := &file_cloud_proto_msgTypes[4] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -311,7 +469,7 @@ func (x *ListToolsRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use ListToolsRequest.ProtoReflect.Descriptor instead. func (*ListToolsRequest) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{2} + return file_cloud_proto_rawDescGZIP(), []int{4} } type ListToolsResponse struct { @@ -324,7 +482,7 @@ type ListToolsResponse struct { func (x *ListToolsResponse) Reset() { *x = ListToolsResponse{} - mi := &file_cloud_proto_msgTypes[3] + mi := &file_cloud_proto_msgTypes[5] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -336,7 +494,7 @@ func (x *ListToolsResponse) String() string { func (*ListToolsResponse) ProtoMessage() {} func (x *ListToolsResponse) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[3] + mi := &file_cloud_proto_msgTypes[5] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -349,7 +507,7 @@ func (x *ListToolsResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use ListToolsResponse.ProtoReflect.Descriptor instead. func (*ListToolsResponse) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{3} + return file_cloud_proto_rawDescGZIP(), []int{5} } func (x *ListToolsResponse) GetTools() []*ToolDefinition { @@ -391,7 +549,7 @@ type ToolDefinition struct { func (x *ToolDefinition) Reset() { *x = ToolDefinition{} - mi := &file_cloud_proto_msgTypes[4] + mi := &file_cloud_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -403,7 +561,7 @@ func (x *ToolDefinition) String() string { func (*ToolDefinition) ProtoMessage() {} func (x *ToolDefinition) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[4] + mi := &file_cloud_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -416,7 +574,7 @@ func (x *ToolDefinition) ProtoReflect() protoreflect.Message { // Deprecated: Use ToolDefinition.ProtoReflect.Descriptor instead. func (*ToolDefinition) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{4} + return file_cloud_proto_rawDescGZIP(), []int{6} } func (x *ToolDefinition) GetName() string { @@ -464,7 +622,7 @@ type CallToolRequest struct { func (x *CallToolRequest) Reset() { *x = CallToolRequest{} - mi := &file_cloud_proto_msgTypes[5] + mi := &file_cloud_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -476,7 +634,7 @@ func (x *CallToolRequest) String() string { func (*CallToolRequest) ProtoMessage() {} func (x *CallToolRequest) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[5] + mi := &file_cloud_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -489,7 +647,7 @@ func (x *CallToolRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use CallToolRequest.ProtoReflect.Descriptor instead. func (*CallToolRequest) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{5} + return file_cloud_proto_rawDescGZIP(), []int{7} } func (x *CallToolRequest) GetName() string { @@ -529,7 +687,7 @@ type CallToolResponse struct { func (x *CallToolResponse) Reset() { *x = CallToolResponse{} - mi := &file_cloud_proto_msgTypes[6] + mi := &file_cloud_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -541,7 +699,7 @@ func (x *CallToolResponse) String() string { func (*CallToolResponse) ProtoMessage() {} func (x *CallToolResponse) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[6] + mi := &file_cloud_proto_msgTypes[8] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -554,7 +712,7 @@ func (x *CallToolResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use CallToolResponse.ProtoReflect.Descriptor instead. func (*CallToolResponse) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{6} + return file_cloud_proto_rawDescGZIP(), []int{8} } func (x *CallToolResponse) GetSuccess() bool { @@ -725,7 +883,7 @@ type CancelStreamRequest struct { func (x *CancelStreamRequest) Reset() { *x = CancelStreamRequest{} - mi := &file_cloud_proto_msgTypes[7] + mi := &file_cloud_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -737,7 +895,7 @@ func (x *CancelStreamRequest) String() string { func (*CancelStreamRequest) ProtoMessage() {} func (x *CancelStreamRequest) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[7] + mi := &file_cloud_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -750,7 +908,7 @@ func (x *CancelStreamRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use CancelStreamRequest.ProtoReflect.Descriptor instead. func (*CancelStreamRequest) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{7} + return file_cloud_proto_rawDescGZIP(), []int{9} } func (x *CancelStreamRequest) GetStreamRequestId() string { @@ -777,7 +935,7 @@ type HostInfo struct { func (x *HostInfo) Reset() { *x = HostInfo{} - mi := &file_cloud_proto_msgTypes[8] + mi := &file_cloud_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -789,7 +947,7 @@ func (x *HostInfo) String() string { func (*HostInfo) ProtoMessage() {} func (x *HostInfo) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[8] + mi := &file_cloud_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -802,7 +960,7 @@ func (x *HostInfo) ProtoReflect() protoreflect.Message { // Deprecated: Use HostInfo.ProtoReflect.Descriptor instead. func (*HostInfo) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{8} + return file_cloud_proto_rawDescGZIP(), []int{10} } func (x *HostInfo) GetId() string { @@ -870,7 +1028,7 @@ type ListHostsResult struct { func (x *ListHostsResult) Reset() { *x = ListHostsResult{} - mi := &file_cloud_proto_msgTypes[9] + mi := &file_cloud_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -882,7 +1040,7 @@ func (x *ListHostsResult) String() string { func (*ListHostsResult) ProtoMessage() {} func (x *ListHostsResult) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[9] + mi := &file_cloud_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -895,7 +1053,7 @@ func (x *ListHostsResult) ProtoReflect() protoreflect.Message { // Deprecated: Use ListHostsResult.ProtoReflect.Descriptor instead. func (*ListHostsResult) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{9} + return file_cloud_proto_rawDescGZIP(), []int{11} } func (x *ListHostsResult) GetHosts() []*HostInfo { @@ -926,7 +1084,7 @@ type ContainerInfo struct { func (x *ContainerInfo) Reset() { *x = ContainerInfo{} - mi := &file_cloud_proto_msgTypes[10] + mi := &file_cloud_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -938,7 +1096,7 @@ func (x *ContainerInfo) String() string { func (*ContainerInfo) ProtoMessage() {} func (x *ContainerInfo) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[10] + mi := &file_cloud_proto_msgTypes[12] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -951,7 +1109,7 @@ func (x *ContainerInfo) ProtoReflect() protoreflect.Message { // Deprecated: Use ContainerInfo.ProtoReflect.Descriptor instead. func (*ContainerInfo) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{10} + return file_cloud_proto_rawDescGZIP(), []int{12} } func (x *ContainerInfo) GetId() string { @@ -1047,7 +1205,7 @@ type ListContainersResult struct { func (x *ListContainersResult) Reset() { *x = ListContainersResult{} - mi := &file_cloud_proto_msgTypes[11] + mi := &file_cloud_proto_msgTypes[13] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1059,7 +1217,7 @@ func (x *ListContainersResult) String() string { func (*ListContainersResult) ProtoMessage() {} func (x *ListContainersResult) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[11] + mi := &file_cloud_proto_msgTypes[13] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1072,7 +1230,7 @@ func (x *ListContainersResult) ProtoReflect() protoreflect.Message { // Deprecated: Use ListContainersResult.ProtoReflect.Descriptor instead. func (*ListContainersResult) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{11} + return file_cloud_proto_rawDescGZIP(), []int{13} } func (x *ListContainersResult) GetContainers() []*ContainerInfo { @@ -1104,7 +1262,7 @@ type ContainerStatEntry struct { func (x *ContainerStatEntry) Reset() { *x = ContainerStatEntry{} - mi := &file_cloud_proto_msgTypes[12] + mi := &file_cloud_proto_msgTypes[14] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1116,7 +1274,7 @@ func (x *ContainerStatEntry) String() string { func (*ContainerStatEntry) ProtoMessage() {} func (x *ContainerStatEntry) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[12] + mi := &file_cloud_proto_msgTypes[14] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1129,7 +1287,7 @@ func (x *ContainerStatEntry) ProtoReflect() protoreflect.Message { // Deprecated: Use ContainerStatEntry.ProtoReflect.Descriptor instead. func (*ContainerStatEntry) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{12} + return file_cloud_proto_rawDescGZIP(), []int{14} } func (x *ContainerStatEntry) GetId() string { @@ -1232,7 +1390,7 @@ type ContainerStatsResult struct { func (x *ContainerStatsResult) Reset() { *x = ContainerStatsResult{} - mi := &file_cloud_proto_msgTypes[13] + mi := &file_cloud_proto_msgTypes[15] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1244,7 +1402,7 @@ func (x *ContainerStatsResult) String() string { func (*ContainerStatsResult) ProtoMessage() {} func (x *ContainerStatsResult) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[13] + mi := &file_cloud_proto_msgTypes[15] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1257,7 +1415,7 @@ func (x *ContainerStatsResult) ProtoReflect() protoreflect.Message { // Deprecated: Use ContainerStatsResult.ProtoReflect.Descriptor instead. func (*ContainerStatsResult) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{13} + return file_cloud_proto_rawDescGZIP(), []int{15} } func (x *ContainerStatsResult) GetStats() []*ContainerStatEntry { @@ -1280,7 +1438,7 @@ type LogEntry struct { func (x *LogEntry) Reset() { *x = LogEntry{} - mi := &file_cloud_proto_msgTypes[14] + mi := &file_cloud_proto_msgTypes[16] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1292,7 +1450,7 @@ func (x *LogEntry) String() string { func (*LogEntry) ProtoMessage() {} func (x *LogEntry) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[14] + mi := &file_cloud_proto_msgTypes[16] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1305,7 +1463,7 @@ func (x *LogEntry) ProtoReflect() protoreflect.Message { // Deprecated: Use LogEntry.ProtoReflect.Descriptor instead. func (*LogEntry) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{14} + return file_cloud_proto_rawDescGZIP(), []int{16} } func (x *LogEntry) GetTimestamp() int64 { @@ -1346,7 +1504,7 @@ type FetchLogsResult struct { func (x *FetchLogsResult) Reset() { *x = FetchLogsResult{} - mi := &file_cloud_proto_msgTypes[15] + mi := &file_cloud_proto_msgTypes[17] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1358,7 +1516,7 @@ func (x *FetchLogsResult) String() string { func (*FetchLogsResult) ProtoMessage() {} func (x *FetchLogsResult) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[15] + mi := &file_cloud_proto_msgTypes[17] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1371,7 +1529,7 @@ func (x *FetchLogsResult) ProtoReflect() protoreflect.Message { // Deprecated: Use FetchLogsResult.ProtoReflect.Descriptor instead. func (*FetchLogsResult) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{15} + return file_cloud_proto_rawDescGZIP(), []int{17} } func (x *FetchLogsResult) GetContainerName() string { @@ -1415,7 +1573,7 @@ type InspectContainerResult struct { func (x *InspectContainerResult) Reset() { *x = InspectContainerResult{} - mi := &file_cloud_proto_msgTypes[16] + mi := &file_cloud_proto_msgTypes[18] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1427,7 +1585,7 @@ func (x *InspectContainerResult) String() string { func (*InspectContainerResult) ProtoMessage() {} func (x *InspectContainerResult) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[16] + mi := &file_cloud_proto_msgTypes[18] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1440,7 +1598,7 @@ func (x *InspectContainerResult) ProtoReflect() protoreflect.Message { // Deprecated: Use InspectContainerResult.ProtoReflect.Descriptor instead. func (*InspectContainerResult) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{16} + return file_cloud_proto_rawDescGZIP(), []int{18} } func (x *InspectContainerResult) GetId() string { @@ -1582,7 +1740,7 @@ type ActionResult struct { func (x *ActionResult) Reset() { *x = ActionResult{} - mi := &file_cloud_proto_msgTypes[17] + mi := &file_cloud_proto_msgTypes[19] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1594,7 +1752,7 @@ func (x *ActionResult) String() string { func (*ActionResult) ProtoMessage() {} func (x *ActionResult) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[17] + mi := &file_cloud_proto_msgTypes[19] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1607,7 +1765,7 @@ func (x *ActionResult) ProtoReflect() protoreflect.Message { // Deprecated: Use ActionResult.ProtoReflect.Descriptor instead. func (*ActionResult) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{17} + return file_cloud_proto_rawDescGZIP(), []int{19} } func (x *ActionResult) GetSuccess() bool { @@ -1650,7 +1808,7 @@ type DeployResult struct { func (x *DeployResult) Reset() { *x = DeployResult{} - mi := &file_cloud_proto_msgTypes[18] + mi := &file_cloud_proto_msgTypes[20] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1662,7 +1820,7 @@ func (x *DeployResult) String() string { func (*DeployResult) ProtoMessage() {} func (x *DeployResult) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[18] + mi := &file_cloud_proto_msgTypes[20] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1675,7 +1833,7 @@ func (x *DeployResult) ProtoReflect() protoreflect.Message { // Deprecated: Use DeployResult.ProtoReflect.Descriptor instead. func (*DeployResult) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{18} + return file_cloud_proto_rawDescGZIP(), []int{20} } func (x *DeployResult) GetSuccess() bool { @@ -1710,7 +1868,7 @@ type NotificationResult struct { func (x *NotificationResult) Reset() { *x = NotificationResult{} - mi := &file_cloud_proto_msgTypes[19] + mi := &file_cloud_proto_msgTypes[21] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1722,7 +1880,7 @@ func (x *NotificationResult) String() string { func (*NotificationResult) ProtoMessage() {} func (x *NotificationResult) ProtoReflect() protoreflect.Message { - mi := &file_cloud_proto_msgTypes[19] + mi := &file_cloud_proto_msgTypes[21] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1735,7 +1893,7 @@ func (x *NotificationResult) ProtoReflect() protoreflect.Message { // Deprecated: Use NotificationResult.ProtoReflect.Descriptor instead. func (*NotificationResult) Descriptor() ([]byte, []int) { - return file_cloud_proto_rawDescGZIP(), []int{19} + return file_cloud_proto_rawDescGZIP(), []int{21} } func (x *NotificationResult) GetSuccess() bool { @@ -1764,14 +1922,25 @@ const file_cloud_proto_rawDesc = "" + "list_tools\x18\x02 \x01(\v2\x17.cloud.ListToolsRequestH\x00R\tlistTools\x125\n" + "\tcall_tool\x18\x03 \x01(\v2\x16.cloud.CallToolRequestH\x00R\bcallTool\x12A\n" + "\rcancel_stream\x18\x04 \x01(\v2\x1a.cloud.CancelStreamRequestH\x00R\fcancelStreamB\x06\n" + - "\x04type\"\xa8\x01\n" + + "\x04type\"\xd8\x01\n" + "\fToolResponse\x12\x1d\n" + "\n" + "request_id\x18\x01 \x01(\tR\trequestId\x129\n" + "\n" + "list_tools\x18\x02 \x01(\v2\x18.cloud.ListToolsResponseH\x00R\tlistTools\x126\n" + - "\tcall_tool\x18\x03 \x01(\v2\x17.cloud.CallToolResponseH\x00R\bcallToolB\x06\n" + - "\x04type\"\x12\n" + + "\tcall_tool\x18\x03 \x01(\v2\x17.cloud.CallToolResponseH\x00R\bcallTool\x12.\n" + + "\tlog_batch\x18\x04 \x01(\v2\x0f.cloud.LogBatchH\x00R\blogBatchB\x06\n" + + "\x04type\":\n" + + "\bLogBatch\x12.\n" + + "\aentries\x18\x01 \x03(\v2\x14.cloud.LogBatchEntryR\aentries\"\xdd\x01\n" + + "\rLogBatchEntry\x12\x17\n" + + "\ahost_id\x18\x01 \x01(\tR\x06hostId\x12!\n" + + "\fcontainer_id\x18\x02 \x01(\tR\vcontainerId\x12%\n" + + "\x0econtainer_name\x18\x03 \x01(\tR\rcontainerName\x12!\n" + + "\ftimestamp_ns\x18\x04 \x01(\x03R\vtimestampNs\x12\x18\n" + + "\amessage\x18\x05 \x01(\tR\amessage\x12\x16\n" + + "\x06stream\x18\x06 \x01(\tR\x06stream\x12\x14\n" + + "\x05level\x18\a \x01(\tR\x05level\"\x12\n" + "\x10ListToolsRequest\"Z\n" + "\x11ListToolsResponse\x12+\n" + "\x05tools\x18\x01 \x03(\v2\x15.cloud.ToolDefinitionR\x05tools\x12\x18\n" + @@ -1922,59 +2091,63 @@ func file_cloud_proto_rawDescGZIP() []byte { } var file_cloud_proto_enumTypes = make([]protoimpl.EnumInfo, 1) -var file_cloud_proto_msgTypes = make([]protoimpl.MessageInfo, 21) +var file_cloud_proto_msgTypes = make([]protoimpl.MessageInfo, 23) var file_cloud_proto_goTypes = []any{ (ToolScope)(0), // 0: cloud.ToolScope (*ToolRequest)(nil), // 1: cloud.ToolRequest (*ToolResponse)(nil), // 2: cloud.ToolResponse - (*ListToolsRequest)(nil), // 3: cloud.ListToolsRequest - (*ListToolsResponse)(nil), // 4: cloud.ListToolsResponse - (*ToolDefinition)(nil), // 5: cloud.ToolDefinition - (*CallToolRequest)(nil), // 6: cloud.CallToolRequest - (*CallToolResponse)(nil), // 7: cloud.CallToolResponse - (*CancelStreamRequest)(nil), // 8: cloud.CancelStreamRequest - (*HostInfo)(nil), // 9: cloud.HostInfo - (*ListHostsResult)(nil), // 10: cloud.ListHostsResult - (*ContainerInfo)(nil), // 11: cloud.ContainerInfo - (*ListContainersResult)(nil), // 12: cloud.ListContainersResult - (*ContainerStatEntry)(nil), // 13: cloud.ContainerStatEntry - (*ContainerStatsResult)(nil), // 14: cloud.ContainerStatsResult - (*LogEntry)(nil), // 15: cloud.LogEntry - (*FetchLogsResult)(nil), // 16: cloud.FetchLogsResult - (*InspectContainerResult)(nil), // 17: cloud.InspectContainerResult - (*ActionResult)(nil), // 18: cloud.ActionResult - (*DeployResult)(nil), // 19: cloud.DeployResult - (*NotificationResult)(nil), // 20: cloud.NotificationResult - nil, // 21: cloud.InspectContainerResult.LabelsEntry + (*LogBatch)(nil), // 3: cloud.LogBatch + (*LogBatchEntry)(nil), // 4: cloud.LogBatchEntry + (*ListToolsRequest)(nil), // 5: cloud.ListToolsRequest + (*ListToolsResponse)(nil), // 6: cloud.ListToolsResponse + (*ToolDefinition)(nil), // 7: cloud.ToolDefinition + (*CallToolRequest)(nil), // 8: cloud.CallToolRequest + (*CallToolResponse)(nil), // 9: cloud.CallToolResponse + (*CancelStreamRequest)(nil), // 10: cloud.CancelStreamRequest + (*HostInfo)(nil), // 11: cloud.HostInfo + (*ListHostsResult)(nil), // 12: cloud.ListHostsResult + (*ContainerInfo)(nil), // 13: cloud.ContainerInfo + (*ListContainersResult)(nil), // 14: cloud.ListContainersResult + (*ContainerStatEntry)(nil), // 15: cloud.ContainerStatEntry + (*ContainerStatsResult)(nil), // 16: cloud.ContainerStatsResult + (*LogEntry)(nil), // 17: cloud.LogEntry + (*FetchLogsResult)(nil), // 18: cloud.FetchLogsResult + (*InspectContainerResult)(nil), // 19: cloud.InspectContainerResult + (*ActionResult)(nil), // 20: cloud.ActionResult + (*DeployResult)(nil), // 21: cloud.DeployResult + (*NotificationResult)(nil), // 22: cloud.NotificationResult + nil, // 23: cloud.InspectContainerResult.LabelsEntry } var file_cloud_proto_depIdxs = []int32{ - 3, // 0: cloud.ToolRequest.list_tools:type_name -> cloud.ListToolsRequest - 6, // 1: cloud.ToolRequest.call_tool:type_name -> cloud.CallToolRequest - 8, // 2: cloud.ToolRequest.cancel_stream:type_name -> cloud.CancelStreamRequest - 4, // 3: cloud.ToolResponse.list_tools:type_name -> cloud.ListToolsResponse - 7, // 4: cloud.ToolResponse.call_tool:type_name -> cloud.CallToolResponse - 5, // 5: cloud.ListToolsResponse.tools:type_name -> cloud.ToolDefinition - 0, // 6: cloud.ToolDefinition.scope:type_name -> cloud.ToolScope - 10, // 7: cloud.CallToolResponse.list_hosts:type_name -> cloud.ListHostsResult - 12, // 8: cloud.CallToolResponse.list_containers:type_name -> cloud.ListContainersResult - 14, // 9: cloud.CallToolResponse.container_stats:type_name -> cloud.ContainerStatsResult - 18, // 10: cloud.CallToolResponse.action:type_name -> cloud.ActionResult - 16, // 11: cloud.CallToolResponse.fetch_logs:type_name -> cloud.FetchLogsResult - 17, // 12: cloud.CallToolResponse.inspect_container:type_name -> cloud.InspectContainerResult - 19, // 13: cloud.CallToolResponse.deploy:type_name -> cloud.DeployResult - 20, // 14: cloud.CallToolResponse.notification:type_name -> cloud.NotificationResult - 9, // 15: cloud.ListHostsResult.hosts:type_name -> cloud.HostInfo - 11, // 16: cloud.ListContainersResult.containers:type_name -> cloud.ContainerInfo - 13, // 17: cloud.ContainerStatsResult.stats:type_name -> cloud.ContainerStatEntry - 15, // 18: cloud.FetchLogsResult.entries:type_name -> cloud.LogEntry - 21, // 19: cloud.InspectContainerResult.labels:type_name -> cloud.InspectContainerResult.LabelsEntry - 2, // 20: cloud.CloudToolService.ToolStream:input_type -> cloud.ToolResponse - 1, // 21: cloud.CloudToolService.ToolStream:output_type -> cloud.ToolRequest - 21, // [21:22] is the sub-list for method output_type - 20, // [20:21] is the sub-list for method input_type - 20, // [20:20] is the sub-list for extension type_name - 20, // [20:20] is the sub-list for extension extendee - 0, // [0:20] is the sub-list for field type_name + 5, // 0: cloud.ToolRequest.list_tools:type_name -> cloud.ListToolsRequest + 8, // 1: cloud.ToolRequest.call_tool:type_name -> cloud.CallToolRequest + 10, // 2: cloud.ToolRequest.cancel_stream:type_name -> cloud.CancelStreamRequest + 6, // 3: cloud.ToolResponse.list_tools:type_name -> cloud.ListToolsResponse + 9, // 4: cloud.ToolResponse.call_tool:type_name -> cloud.CallToolResponse + 3, // 5: cloud.ToolResponse.log_batch:type_name -> cloud.LogBatch + 4, // 6: cloud.LogBatch.entries:type_name -> cloud.LogBatchEntry + 7, // 7: cloud.ListToolsResponse.tools:type_name -> cloud.ToolDefinition + 0, // 8: cloud.ToolDefinition.scope:type_name -> cloud.ToolScope + 12, // 9: cloud.CallToolResponse.list_hosts:type_name -> cloud.ListHostsResult + 14, // 10: cloud.CallToolResponse.list_containers:type_name -> cloud.ListContainersResult + 16, // 11: cloud.CallToolResponse.container_stats:type_name -> cloud.ContainerStatsResult + 20, // 12: cloud.CallToolResponse.action:type_name -> cloud.ActionResult + 18, // 13: cloud.CallToolResponse.fetch_logs:type_name -> cloud.FetchLogsResult + 19, // 14: cloud.CallToolResponse.inspect_container:type_name -> cloud.InspectContainerResult + 21, // 15: cloud.CallToolResponse.deploy:type_name -> cloud.DeployResult + 22, // 16: cloud.CallToolResponse.notification:type_name -> cloud.NotificationResult + 11, // 17: cloud.ListHostsResult.hosts:type_name -> cloud.HostInfo + 13, // 18: cloud.ListContainersResult.containers:type_name -> cloud.ContainerInfo + 15, // 19: cloud.ContainerStatsResult.stats:type_name -> cloud.ContainerStatEntry + 17, // 20: cloud.FetchLogsResult.entries:type_name -> cloud.LogEntry + 23, // 21: cloud.InspectContainerResult.labels:type_name -> cloud.InspectContainerResult.LabelsEntry + 2, // 22: cloud.CloudToolService.ToolStream:input_type -> cloud.ToolResponse + 1, // 23: cloud.CloudToolService.ToolStream:output_type -> cloud.ToolRequest + 23, // [23:24] is the sub-list for method output_type + 22, // [22:23] is the sub-list for method input_type + 22, // [22:22] is the sub-list for extension type_name + 22, // [22:22] is the sub-list for extension extendee + 0, // [0:22] is the sub-list for field type_name } func init() { file_cloud_proto_init() } @@ -1990,8 +2163,9 @@ func file_cloud_proto_init() { file_cloud_proto_msgTypes[1].OneofWrappers = []any{ (*ToolResponse_ListTools)(nil), (*ToolResponse_CallTool)(nil), + (*ToolResponse_LogBatch)(nil), } - file_cloud_proto_msgTypes[6].OneofWrappers = []any{ + file_cloud_proto_msgTypes[8].OneofWrappers = []any{ (*CallToolResponse_ListHosts)(nil), (*CallToolResponse_ListContainers)(nil), (*CallToolResponse_ContainerStats)(nil), @@ -2007,7 +2181,7 @@ func file_cloud_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_cloud_proto_rawDesc), len(file_cloud_proto_rawDesc)), NumEnums: 1, - NumMessages: 21, + NumMessages: 23, NumExtensions: 0, NumServices: 1, }, diff --git a/protos/cloud.proto b/protos/cloud.proto index 460537ab..d961d64c 100644 --- a/protos/cloud.proto +++ b/protos/cloud.proto @@ -23,9 +23,30 @@ message ToolResponse { oneof type { ListToolsResponse list_tools = 2; CallToolResponse call_tool = 3; + // Unsolicited server-push: batched container log lines streamed from + // Dozzle to Cloud for ingestion into VictoriaLogs. request_id is empty + // for log batches — they are not replies to a ToolRequest. + LogBatch log_batch = 4; } } +// Batch of log lines from one or more containers. Dozzle pushes these +// continuously while connected. Cloud routes them to VictoriaLogs scoped +// to the owning user (derived from the connection's auth). +message LogBatch { + repeated LogBatchEntry entries = 1; +} + +message LogBatchEntry { + string host_id = 1; + string container_id = 2; + string container_name = 3; + int64 timestamp_ns = 4; // unix nanoseconds + string message = 5; + string stream = 6; // "stdout" or "stderr" + string level = 7; // "info", "warn", "error", etc. (best-effort) +} + message ListToolsRequest {} message ListToolsResponse {