WIP: fs_scrub controller

Signed-off-by: Dmitry Sharshakov <[email protected]>

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

@@ -0,0 +1,180 @@
+// 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"
+	"math/rand/v2"
+	"time"
+
+	"github.com/cosi-project/runtime/pkg/controller"
+	"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/internal/app/machined/pkg/system/events"
+	"github.com/siderolabs/talos/internal/app/machined/pkg/system/runner"
+	"github.com/siderolabs/talos/internal/app/machined/pkg/system/runner/process"
+	"github.com/siderolabs/talos/internal/pkg/environment"
+	"github.com/siderolabs/talos/pkg/machinery/constants"
+	"github.com/siderolabs/talos/pkg/machinery/resources/block"
+)
+
+type scrubSchedule struct {
+	period   time.Duration
+	upcoming time.Time
+}
+
+// FSScrubController watches v1alpha1.Config and schedules filesystem online check tasks.
+type FSScrubController struct {
+	Runtime  runtime.Runtime
+	schedule map[string]scrubSchedule
+}
+
+// Name implements controller.Controller interface.
+func (ctrl *FSScrubController) Name() string {
+	return "runtime.FSScrubController"
+}
+
+// Inputs implements controller.Controller interface.
+func (ctrl *FSScrubController) Inputs() []controller.Input {
+	return []controller.Input{
+		{
+			Namespace: block.NamespaceName,
+			Type:      block.VolumeStatusType,
+			Kind:      controller.InputWeak,
+		},
+		{
+			Namespace: block.NamespaceName,
+			Type:      block.VolumeConfigType,
+			Kind:      controller.InputWeak,
+		},
+	}
+}
+
+// Outputs implements controller.Controller interface.
+func (ctrl *FSScrubController) Outputs() []controller.Output {
+	return []controller.Output{}
+}
+
+// Run implements controller.Controller interface.
+func (ctrl *FSScrubController) Run(ctx context.Context, r controller.Runtime, logger *zap.Logger) error {
+	var (
+		ticker  *time.Ticker
+		tickerC <-chan time.Time
+	)
+
+	tickerStop := func() {
+		if ticker == nil {
+			return
+		}
+
+		ticker.Stop()
+
+		ticker = nil
+		tickerC = nil
+	}
+
+	defer tickerStop()
+
+	tickerStop()
+
+	ticker = time.NewTicker(15 * time.Second)
+	tickerC = ticker.C
+
+	ctrl.schedule = make(map[string]scrubSchedule)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-tickerC:
+			if err := ctrl.scrub("/var", []string{}); err != nil {
+				return fmt.Errorf("error running filesystem scrub: %w", err)
+			}
+
+			continue
+		case <-r.EventCh():
+			err := ctrl.updateSchedule(ctx, r, logger)
+			if err != nil {
+				return err
+			}
+		}
+	}
+}
+
+func (ctrl *FSScrubController) updateSchedule(ctx context.Context, r controller.Runtime, logger *zap.Logger) error {
+	volumesStatus, err := safe.ReaderListAll[*block.VolumeStatus](ctx, r)
+	if err != nil && !state.IsNotFoundError(err) {
+		return fmt.Errorf("error getting volume status: %w", err)
+	}
+
+	logger.Warn("reading volume status")
+	volumesStatus.ForEach(func(item *block.VolumeStatus) {
+		vol := item.TypedSpec()
+
+		logger.Warn("volume status", zap.Reflect("item", vol))
+
+		if vol.Phase != block.VolumePhaseReady {
+			logger.Warn("vol.Phase != block.VolumePhaseReady", zap.Reflect("item", vol))
+
+			return
+		}
+
+		if vol.Filesystem != block.FilesystemTypeXFS {
+			logger.Warn("vol.Filesystem != block.FilesystemTypeXFS", zap.Reflect("item", vol))
+
+			return
+		}
+
+		volumeConfig, err := safe.ReaderGetByID[*block.VolumeConfig](ctx, r, item.Metadata().ID())
+		if err != nil {
+			logger.Warn("err", zap.Error(err))
+
+			return
+		}
+
+		mountpoint := volumeConfig.TypedSpec().Mount.TargetPath
+
+		if _, ok := ctrl.schedule[mountpoint]; !ok {
+			per := 10 * time.Second
+			seconds := rand.Int64N(int64(per.Seconds()))
+
+			ctrl.schedule[mountpoint] = scrubSchedule{
+				period:   per,
+				upcoming: time.Now().Add(time.Duration(seconds * int64(time.Second))),
+			}
+
+			logger.Warn("scheduled", zap.String("path", mountpoint), zap.Reflect("upcoming", ctrl.schedule[mountpoint].upcoming))
+		}
+	})
+
+	return nil
+}
+
+func (ctrl *FSScrubController) scrub(mountpoint string, opts []string) error {
+	args := []string{"/usr/sbin/xfs_scrub", "-T", "-v"}
+	args = append(args, opts...)
+	args = append(args, mountpoint)
+
+	r := process.NewRunner(
+		false,
+		&runner.Args{
+			ID:          "fs_scrub",
+			ProcessArgs: args,
+		},
+		runner.WithLoggingManager(ctrl.Runtime.Logging()),
+		runner.WithEnv(environment.Get(ctrl.Runtime.Config())),
+		runner.WithOOMScoreAdj(-999),
+		runner.WithDroppedCapabilities(constants.XFSScrubDroppedCapabilities),
+		runner.WithPriority(19),
+		runner.WithIOPriority(runner.IoprioClassIdle, 7),
+		runner.WithSchedulingPolicy(runner.SchedulingPolicyIdle),
+	)
+
+	return r.Run(func(s events.ServiceState, msg string, args ...any) {})
+}

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

@@ -339,6 +339,9 @@ func (ctrl *Controller) Run(ctx context.Context, drainer *runtime.Drainer) error
 		runtimecontrollers.NewUniqueMachineTokenController(),
 		&runtimecontrollers.WatchdogTimerConfigController{},
 		&runtimecontrollers.WatchdogTimerController{},
+		&runtimecontrollers.FSScrubController{
+			Runtime: ctrl.v1alpha1Runtime,
+		},
 		&secrets.APICertSANsController{},
 		&secrets.APIController{},
 		&secrets.EtcdController{},

pkg/machinery/constants/constants.go

@@ -1273,6 +1273,38 @@ var DefaultDroppedCapabilities = map[string]struct{}{
 	"cap_sys_module": {},
 }
 
+// XFSScrubDroppedCapabilities is the set of capabilities to drop for xfs_scrub.
+// All but cap_sys_admin cap_fowner cap_dac_override cap_dac_read_search cap_sys_rawio
+var XFSScrubDroppedCapabilities = map[string]struct{}{
+	"cap_audit_control":    {},
+	"cap_audit_write":      {},
+	"cap_chown":            {},
+	"cap_fsetid":           {},
+	"cap_ipc_lock":         {},
+	"cap_ipc_owner":        {},
+	"cap_kill":             {},
+	"cap_lease":            {},
+	"cap_linux_immutable":  {},
+	"cap_mknod":            {},
+	"cap_net_admin":        {},
+	"cap_net_bind_service": {},
+	"cap_net_broadcast":    {},
+	"cap_net_raw":          {},
+	"cap_setfcap":          {},
+	"cap_setgid":           {},
+	"cap_setpcap":          {},
+	"cap_setuid":           {},
+	"cap_sys_boot":         {},
+	"cap_sys_chroot":       {},
+	"cap_sys_module":       {},
+	"cap_sys_nice":         {},
+	"cap_sys_pacct":        {},
+	"cap_sys_ptrace":       {},
+	"cap_sys_resource":     {},
+	"cap_sys_time":         {},
+	"cap_sys_tty_config":   {},
+}
+
 // UdevdDroppedCapabilities is the set of capabilities to drop for udevd.
 var UdevdDroppedCapabilities = map[string]struct{}{
 	"cap_sys_boot": {},