fixes to only running

internal/docker/client.go

@@ -66,7 +66,6 @@ func NewClient(cli DockerCLI, host container.Host) *DockerClient {
 	}
 }
 
-// NewClientWithFilters creates a new instance of Client with docker filters
 func NewLocalClient(hostname string) (*DockerClient, error) {
 	cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation(), client.WithUserAgent("Docker-Client/Dozzle"))
 

internal/ingest/victoria.go

@@ -0,0 +1,112 @@
+package ingest
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/amir20/dozzle/internal/container"
+	container_support "github.com/amir20/dozzle/internal/support/container"
+	"github.com/rs/zerolog/log"
+)
+
+type VictoriaIngestor struct {
+	service container_support.ClientService
+	logs    chan *container.LogEvent
+}
+
+func NewVictoriaIngestor(service container_support.ClientService) *VictoriaIngestor {
+	return &VictoriaIngestor{
+		service: service,
+		logs:    make(chan *container.LogEvent),
+	}
+}
+
+func (v *VictoriaIngestor) consumeLogs(ctx context.Context) error {
+	pr, pw := io.Pipe()
+
+	req, err := http.NewRequestWithContext(ctx, "POST", "http://localhost:9428/insert/jsonline?_stream_fields=c&_time_field=ts&_msg_field=m", pr)
+	if err != nil {
+		return err
+	}
+
+	// Set headers as needed.
+	req.Header.Set("Content-Type", "application/stream+json")
+
+	go func() {
+		defer pw.Close()
+		writer := json.NewEncoder(pw)
+		for event := range v.logs {
+			log.Debug().Interface("event", event).Msg("Writing log to Victoria")
+			err := writer.Encode(event)
+			if err != nil {
+				log.Error().Err(err).Msg("Error encoding log event")
+			}
+			log.Debug().Interface("event", event).Msg("Log written to Victoria")
+		}
+	}()
+
+	resp, err := http.DefaultClient.Do(req)
+
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	return nil
+}
+
+func (v *VictoriaIngestor) streamLogs(ctx context.Context, c container.Container) {
+	err := v.service.StreamLogs(ctx, c, time.Now(), container.STDALL, v.logs)
+	if err != nil {
+		if errors.Is(err, io.EOF) {
+			log.Debug().Str("container", c.ID).Msg("streaming ended")
+
+		} else if !errors.Is(err, context.Canceled) {
+			log.Error().Err(err).Str("container", c.ID).Msg("unknown error while streaming logs")
+		}
+	}
+}
+
+func (v *VictoriaIngestor) Start(ctx context.Context) error {
+	go func() {
+		err := v.consumeLogs(ctx)
+		if err != nil {
+			log.Fatal().Err(err).Msg("Error consuming logs")
+		}
+	}()
+
+	go func() {
+		newContainers := make(chan container.Container)
+		defer close(newContainers)
+
+		v.service.SubscribeContainersStarted(ctx, newContainers)
+
+		for {
+			select {
+			case c := <-newContainers:
+				go v.streamLogs(ctx, c)
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	containers, err := v.service.ListContainers(ctx, container.ContainerLabels{})
+	if err != nil {
+		return err
+	}
+
+	for _, c := range containers {
+		if c.State == "running" {
+			go v.streamLogs(ctx, c)
+		}
+	}
+
+	<-ctx.Done()
+
+	return nil
+}

internal/support/cli/args.go

@@ -35,6 +35,7 @@ type Args struct {
 	Generate         *GenerateCmd        `arg:"subcommand:generate" help:"generates a configuration file for simple auth"`
 	Agent            *AgentCmd           `arg:"subcommand:agent" help:"starts the agent"`
 	AgentTest        *AgentTestCmd       `arg:"subcommand:agent-test" help:"tests an agent"`
+	Ingest           *IngestCmd          `arg:"subcommand:ingest" help:"starts the ingest"`
 }
 
 type Runnable interface {

internal/support/cli/ingest_command.go

@@ -0,0 +1,40 @@
+package cli
+
+import (
+	"context"
+	"embed"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/amir20/dozzle/internal/docker"
+	"github.com/amir20/dozzle/internal/ingest"
+	docker_support "github.com/amir20/dozzle/internal/support/docker"
+	"github.com/rs/zerolog/log"
+)
+
+type IngestCmd struct {
+	Addr string `arg:"--ingest-addr,env:DOZZLE_INGEST_ADDR" default:"localhost:9428" help:"sets the host:port to bind for the ingest"`
+}
+
+func (c IngestCmd) Run(args Args, embeddedCerts embed.FS) error {
+	client, err := docker.NewLocalClient(args.Hostname)
+	service := docker_support.NewDockerClientService(client, args.Filter)
+	if err != nil {
+		return err
+	}
+	ingestor := ingest.NewVictoriaIngestor(service)
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+
+	go func() {
+		log.Info().Msgf("Dozzle ingestor version %s", args.Version())
+		if err := ingestor.Start(ctx); err != nil {
+			log.Error().Err(err).Msg("Ingestor failed")
+		}
+	}()
+	<-ctx.Done()
+	log.Info().Msg("Ingestor stopped")
+	stop()
+
+	return nil
+}

main.go

@@ -16,6 +16,7 @@ import (
 	"github.com/amir20/dozzle/internal/agent"
 	"github.com/amir20/dozzle/internal/auth"
 	"github.com/amir20/dozzle/internal/docker"
+
 	"github.com/amir20/dozzle/internal/k8s"
 	"github.com/amir20/dozzle/internal/support/cli"
 	docker_support "github.com/amir20/dozzle/internal/support/docker"
@@ -42,6 +43,7 @@ func main() {
 		err := runnable.Run(args, certs)
 		if err != nil {
 			log.Fatal().Err(err).Msg("Failed to run command")
+
 		}
 
 		os.Exit(0)