Delete container logs older than one day

cmd/icinga-kubernetes/main.go

@@ -198,6 +198,26 @@ func main() {
 			return
 		}
 	}, periodic.Immediate()).Stop()
+
+	defer periodic.Start(ctx, time.Hour, func(tick periodic.Tick) {
+		olderThan := tick.Time.AddDate(0, 0, -1)
+
+		_, err := db.CleanupOlderThan(
+			ctx, database.CleanupStmt{
+				Table:  "container_log",
+				PK:     "container_id, pod_id, period",
+				Column: "last_update",
+			}, 5000, olderThan,
+		)
+		if err != nil {
+			select {
+			case errs <- err:
+			case <-ctx.Done():
+			}
+
+			return
+		}
+	}, periodic.Immediate()).Stop()
 	com.ErrgroupReceive(ctx, g, errs)
 
 	if err := g.Wait(); err != nil {

pkg/schema/v1/container.go

@@ -86,6 +86,7 @@ type ContainerMount struct {
 type ContainerLogMeta struct {
 	Logs       string          `db:"logs"`
 	LastUpdate types.UnixMilli `db:"last_update"`
+	Period     types.UnixMilli
 }
 
 type ContainerLog struct {
@@ -124,7 +125,23 @@ func (cl *ContainerLog) syncContainerLogs(ctx context.Context, clientset *kubern
 	}
 
 	cl.LastUpdate = types.UnixMilli(time.Now())
-	cl.Logs += string(logs)
+	// Calculate period
+	currentTimeMillis := time.Now().UnixMilli()
+	periodStartMillis := currentTimeMillis - (currentTimeMillis % (3600 * 1000))
+	cl.Period = types.UnixMilli(time.UnixMilli(periodStartMillis))
+
+	// Check if logs for the current period already exist
+	existingLog := &ContainerLog{}
+	err = db.Get(existingLog, "SELECT * FROM container_log WHERE container_uuid = ? AND pod_uuid = ? AND period = ?", cl.ContainerUuid, cl.PodUuid, cl.Period)
+	if errors.Is(err, sql.ErrNoRows) {
+		// No existing logs for this period, insert new log
+		cl.Logs = string(logs)
+	} else if err != nil {
+		return err
+	} else {
+		// Existing logs found for this period, concatenate logs
+		cl.Logs += string(logs)
+	}
 	entities := make(chan interface{}, 1)
 	entities <- cl
 	close(entities)
@@ -272,9 +289,13 @@ func SyncContainers(ctx context.Context, db *database.Database, g *errgroup.Grou
 func warmup(ctx context.Context, db *database.Database) error {
 	g, ctx := errgroup.WithContext(ctx)
 
+	query := `SELECT cl.* FROM container_log cl INNER JOIN (SELECT container_uuid, pod_uuid, MAX(last_update) as last_update
+       FROM container_log GROUP BY container_uuid) max_cl ON cl.container_uuid = max_cl.container_uuid AND cl.pod_uuid = max_cl.pod_uuid 
+	   AND cl.last_update = max_cl.last_update`
+
 	entities, errs := db.YieldAll(ctx, func() (interface{}, error) {
 		return &ContainerLog{}, nil
-	}, db.BuildSelectStmt(ContainerLog{}, ContainerLog{}))
+	}, query)
 	com.ErrgroupReceive(ctx, g, errs)
 
 	g.Go(func() error {

schema/mysql/schema.sql

@@ -173,8 +173,9 @@ CREATE TABLE container_log (
   pod_uuid binary(16) NOT NULL,
   logs longtext NOT NULL,
   last_update bigint NOT NULL,
+  `period` bigint unsigned NOT NULL,
 
-  PRIMARY KEY (container_uuid, pod_uuid)
+  PRIMARY KEY (container_uuid, pod_uuid, period)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
 
 CREATE TABLE deployment (