feat: implement logs persistence

internal/app/machined/pkg/controllers/block/volume_config.go

@@ -464,7 +464,7 @@ func (ctrl *VolumeConfigController) manageStandardVolumes(ctx context.Context, r
 	}{
 		// /var/log
 		{
-			Path:         "/var/log",
+			Path:         constants.LogMountPoint,
 			Mode:         0o755,
 			SELinuxLabel: "system_u:object_r:var_log_t:s0",
 		},

internal/app/machined/pkg/controllers/block/volume_config_test.go

@@ -157,7 +157,7 @@ func (suite *VolumeConfigSuite) TestReconcileDefaults() {
 	})
 
 	ctest.AssertResources(suite, []resource.ID{
-		"/var/log",
+		constants.LogMountPoint,
 		"/var/log/audit",
 		"/var/log/containers",
 		"/var/log/pods",

internal/app/machined/pkg/controllers/runtime/log_persistence.go

@@ -0,0 +1,157 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+package runtime
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync/atomic"
+
+	"github.com/cosi-project/runtime/pkg/controller"
+	"github.com/cosi-project/runtime/pkg/resource"
+	"github.com/cosi-project/runtime/pkg/safe"
+	"github.com/cosi-project/runtime/pkg/state"
+	"go.uber.org/zap"
+
+	"github.com/siderolabs/talos/internal/app/machined/pkg/runtime"
+	"github.com/siderolabs/talos/pkg/machinery/constants"
+	"github.com/siderolabs/talos/pkg/machinery/resources/block"
+)
+
+// LogPersistenceController is a controller that persists logs in files.
+type LogPersistenceController struct {
+	V1Alpha1Logging runtime.LoggingManager
+
+	// dummy implementation
+	canLog atomic.Bool
+}
+
+// Name implements controller.Controller interface.
+func (ctrl *LogPersistenceController) Name() string {
+	return "runtime.LogPersistenceController"
+}
+
+// Inputs implements controller.Controller interface.
+func (ctrl *LogPersistenceController) Inputs() []controller.Input {
+	return []controller.Input{
+		{
+			Namespace: block.NamespaceName,
+			Type:      block.VolumeMountStatusType,
+			Kind:      controller.InputStrong,
+		},
+		{
+			Namespace: block.NamespaceName,
+			Type:      block.VolumeMountRequestType,
+			Kind:      controller.InputDestroyReady,
+		},
+	}
+}
+
+// Outputs implements controller.Controller interface.
+func (ctrl *LogPersistenceController) Outputs() []controller.Output {
+	return []controller.Output{
+		{
+			Type: block.VolumeMountRequestType,
+			Kind: controller.OutputShared,
+		},
+	}
+}
+
+func (ctrl *LogPersistenceController) WriteLog(id string, line []byte) error {
+	if !ctrl.canLog.Load() {
+		// logging is not enabled, drop the log line
+		return nil
+	}
+
+	f, err := os.OpenFile(filepath.Join(constants.LogMountPoint, id+".log"), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
+	if err != nil {
+		return fmt.Errorf("error opening log file for %q: %w", id, err)
+	}
+
+	if _, err = f.Write(append(line, '\n')); err != nil {
+		f.Close()
+
+		return fmt.Errorf("error writing log line for %q: %w", id, err)
+	}
+
+	if err = f.Close(); err != nil {
+		return fmt.Errorf("error closing log file for %q: %w", id, err)
+	}
+
+	return nil
+}
+
+func (ctrl *LogPersistenceController) startLogging() {
+	// [TODO]: here we can start logging activities
+	ctrl.canLog.Store(true)
+}
+
+func (ctrl *LogPersistenceController) stopLogging() {
+	// [TODO]: here we should stop all logging activities, close files, flush buffers, etc.
+	// after this call we should not hold /var/log
+	ctrl.canLog.Store(false)
+}
+
+// Run implements controller.Controller interface.
+//
+//nolint:gocyclo
+func (ctrl *LogPersistenceController) Run(ctx context.Context, r controller.Runtime, _ *zap.Logger) error {
+	ctrl.V1Alpha1Logging.SetLineWriter(ctrl)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-r.EventCh():
+		}
+
+		requestID := ctrl.Name() + "-" + constants.LogMountPoint
+
+		// create a volume mount request for the logs volume mount point
+		// to keep it alive and prevent it from being torn down
+		if err := safe.WriterModify(ctx, r,
+			block.NewVolumeMountRequest(block.NamespaceName, requestID),
+			func(v *block.VolumeMountRequest) error {
+				v.TypedSpec().Requester = ctrl.Name()
+				v.TypedSpec().VolumeID = constants.LogMountPoint
+
+				return nil
+			},
+		); err != nil {
+			return fmt.Errorf("error creating volume mount request for user volume mount point: %w", err)
+		}
+
+		vms, err := safe.ReaderGetByID[*block.VolumeMountStatus](ctx, r, requestID)
+		if err != nil {
+			if state.IsNotFoundError(err) {
+				// volume mount not ready yet, wait more
+				continue
+			}
+
+			return fmt.Errorf("error getting volume mount status for log volume: %w", err)
+		}
+
+		switch vms.Metadata().Phase() {
+		case resource.PhaseRunning:
+			if !vms.Metadata().Finalizers().Has(ctrl.Name()) {
+				if err = r.AddFinalizer(ctx, vms.Metadata(), ctrl.Name()); err != nil {
+					return fmt.Errorf("error adding finalizer to volume mount status for log volume: %w", err)
+				}
+
+				ctrl.startLogging()
+			}
+		case resource.PhaseTearingDown:
+			if vms.Metadata().Finalizers().Has(ctrl.Name()) {
+				ctrl.stopLogging()
+
+				if err = r.RemoveFinalizer(ctx, vms.Metadata(), ctrl.Name()); err != nil {
+					return fmt.Errorf("error removing finalizer from volume mount status for log volume: %w", err)
+				}
+			}
+		}
+	}
+}

internal/app/machined/pkg/runtime/logging.go

@@ -24,6 +24,11 @@ type LoggingManager interface {
 	// SetSenders should be thread-safe.
 	SetSenders(senders []LogSender) []LogSender
 
+	// SetLineWriter sets a writer that will receive raw log lines.
+	//
+	// SetLineWriter can be only singularly called, subsequent calls override previous writer.
+	SetLineWriter(w LogWriter)
+
 	// RegisteredLogs returns a list of registered logs containers.
 	RegisteredLogs() []string
 }
@@ -88,3 +93,11 @@ type LogSender interface {
 	// Close should be thread-safe.
 	Close(ctx context.Context) error
 }
+
+// LogWriter provider common interface for text-based log writers.
+type LogWriter interface {
+	// WriteLog writes a single log line.
+	//
+	// WriteLog should be thread-safe.
+	WriteLog(id string, line []byte) error
+}

internal/app/machined/pkg/runtime/logging/circular.go

@@ -49,6 +49,8 @@ type CircularBufferLoggingManager struct {
 	sendersRW      sync.RWMutex
 	senders        []runtime.LogSender
 	sendersChanged chan struct{}
+
+	lineWriter chan runtime.LogWriter
 }
 
 // NewCircularBufferLoggingManager initializes new CircularBufferLoggingManager.
@@ -66,6 +68,7 @@ func NewCircularBufferLoggingManager(fallbackLogger *log.Logger) *CircularBuffer
 		fallbackLogger: fallbackLogger,
 		sendersChanged: make(chan struct{}),
 		compressor:     compressor,
+		lineWriter:     make(chan runtime.LogWriter, 1),
 	}
 }
 
@@ -98,6 +101,16 @@ func (manager *CircularBufferLoggingManager) SetSenders(senders []runtime.LogSen
 	return prevSenders
 }
 
+// SetLineWriter implements runtime.LoggingManager interface.
+func (manager *CircularBufferLoggingManager) SetLineWriter(w runtime.LogWriter) {
+	select {
+	case manager.lineWriter <- w:
+	default:
+		<-manager.lineWriter
+		manager.lineWriter <- w
+	}
+}
+
 // getSenders waits for senders to be set and returns them.
 func (manager *CircularBufferLoggingManager) getSenders() []runtime.LogSender {
 	for {
@@ -193,6 +206,18 @@ func (handler *circularHandler) Writer() (io.WriteCloser, error) {
 					handler.manager.fallbackLogger.Printf("log senders stopped: %s", err)
 				}
 			}()
+
+			go func() {
+				defer func() {
+					if r := recover(); r != nil {
+						handler.manager.fallbackLogger.Printf("log writer panic: %v", r)
+					}
+				}()
+
+				if err := handler.runLineWriter(); err != nil {
+					handler.manager.fallbackLogger.Printf("log writer stopped: %s", err)
+				}
+			}()
 		}
 	}
 
@@ -325,6 +350,29 @@ func (handler *circularHandler) resend(e *runtime.LogEvent) {
 	}
 }
 
+func (handler *circularHandler) runLineWriter() error {
+	r, err := handler.Reader(runtime.WithFollow())
+	if err != nil {
+		return err
+	}
+	defer r.Close() //nolint:errcheck
+
+	w := <-handler.manager.lineWriter
+	handler.manager.lineWriter <- w
+
+	scanner := bufio.NewScanner(r)
+	for scanner.Scan() {
+		l := scanner.Bytes()
+
+		err = w.WriteLog(handler.id, l)
+		if err != nil {
+			return fmt.Errorf("line writer: %w", err)
+		}
+	}
+
+	return fmt.Errorf("scanner: %w", scanner.Err())
+}
+
 // timeStampWriter is a writer that adds a timestamp to each line.
 type timeStampWriter struct {
 	w io.WriteCloser

internal/app/machined/pkg/runtime/logging/file.go

@@ -48,6 +48,9 @@ func (manager *FileLoggingManager) SetSenders([]runtime.LogSender) []runtime.Log
 	return nil
 }
 
+// SetLineWriter implements runtime.LoggingManager interface (by doing nothing).
+func (manager *FileLoggingManager) SetLineWriter(runtime.LogWriter) {}
+
 // RegisteredLogs implements runtime.LoggingManager interface.
 func (manager *FileLoggingManager) RegisteredLogs() []string {
 	var result []string

internal/app/machined/pkg/runtime/logging/null.go

@@ -29,6 +29,9 @@ func (*NullLoggingManager) SetSenders([]runtime.LogSender) []runtime.LogSender {
 	return nil
 }
 
+// SetLineWriter implements runtime.LoggingManager interface (by doing nothing).
+func (*NullLoggingManager) SetLineWriter(runtime.LogWriter) {}
+
 // RegisteredLogs implements runtime.LoggingManager interface (by doing nothing).
 func (*NullLoggingManager) RegisteredLogs() []string {
 	return nil

internal/app/machined/pkg/runtime/v1alpha2/v1alpha2_controller.go

@@ -422,6 +422,9 @@ func (ctrl *Controller) Run(ctx context.Context, drainer *runtime.Drainer) error
 		&runtimecontrollers.KmsgLogDeliveryController{
 			Drainer: drainer,
 		},
+		&runtimecontrollers.LogPersistenceController{
+			V1Alpha1Logging: ctrl.v1alpha1Runtime.Logging(),
+		},
 		&runtimecontrollers.LoadedKernelModuleController{
 			V1Alpha1Mode: ctrl.v1alpha1Runtime.State().Platform().Mode(),
 		},

internal/app/machined/pkg/system/services/kubelet.go

@@ -110,7 +110,7 @@ func (k *Kubelet) Volumes(runtime.Runtime) []string {
 	return []string{
 		"/var/lib",
 		"/var/lib/kubelet",
-		"/var/log",
+		constants.LogMountPoint,
 		"/var/log/audit",
 		"/var/log/containers",
 		"/var/log/pods",

internal/integration/api/selinux.go

@@ -111,7 +111,7 @@ func (suite *SELinuxSuite) TestFileMountLabels() {
 		"/var/lib/cni":                  "system_u:object_r:cni_state_t:s0",
 		"/var/lib/kubelet":              "system_u:object_r:kubelet_state_t:s0",
 		"/var/lib/kubelet/seccomp":      "system_u:object_r:seccomp_profile_t:s0",
-		"/var/log":                      "system_u:object_r:var_log_t:s0",
+		constants.LogMountPoint:         "system_u:object_r:var_log_t:s0",
 		"/var/log/audit":                "system_u:object_r:audit_log_t:s0",
 		constants.KubernetesAuditLogDir: "system_u:object_r:kube_log_t:s0",
 		"/var/log/containers":           "system_u:object_r:containers_log_t:s0",

pkg/machinery/constants/constants.go

@@ -1275,6 +1275,9 @@ const (
 	// UserVolumeMountPoint is the path to the volume mount point for the user volumes.
 	UserVolumeMountPoint = "/var/mnt"
 
+	// LogMountPoint is the path to the logs mount point, and ID of the logs volume.
+	LogMountPoint = "/var/log"
+
 	// UserVolumePrefix is the prefix for the user volumes.
 	UserVolumePrefix = "u-"