lib/controller/trash.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package controller
   6
   7 import (
   8         "context"
   9         "time"
  10
  11         "git.arvados.org/arvados.git/lib/controller/dblock"
  12         "git.arvados.org/arvados.git/sdk/go/auth"
  13         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  14 )
  15
  16 func (h *Handler) periodicWorker(workerName string, interval time.Duration, locker *dblock.DBLocker, run func(context.Context) error) {
  17         logger := ctxlog.FromContext(h.BackgroundContext).WithField("worker", workerName)
  18         ctx := ctxlog.Context(h.BackgroundContext, logger)
  19         if interval <= 0 {
  20                 logger.Debugf("interval is %v, not running worker", interval)
  21                 return
  22         }
  23         locker.Lock(ctx, h.db)
  24         defer locker.Unlock()
  25         for time.Sleep(interval); ctx.Err() == nil; time.Sleep(interval) {
  26                 locker.Check()
  27                 err := run(ctx)
  28                 if err != nil {
  29                         logger.WithError(err).Infof("%s failed", workerName)
  30                 }
  31         }
  32 }
  33
  34 func (h *Handler) trashSweepWorker() {
  35         h.periodicWorker("trash sweep", h.Cluster.Collections.TrashSweepInterval.Duration(), dblock.TrashSweep, func(ctx context.Context) error {
  36                 ctx = auth.NewContext(ctx, &auth.Credentials{Tokens: []string{h.Cluster.SystemRootToken}})
  37                 _, err := h.federation.SysTrashSweep(ctx, struct{}{})
  38                 return err
  39         })
  40 }
  41
  42 func (h *Handler) containerLogSweepWorker() {
  43         h.periodicWorker("container log sweep", h.Cluster.Containers.Logging.SweepInterval.Duration(), dblock.ContainerLogSweep, func(ctx context.Context) error {
  44                 db, err := h.db(ctx)
  45                 if err != nil {
  46                         return err
  47                 }
  48                 res, err := db.ExecContext(ctx, `
  49 DELETE FROM logs
  50  USING containers
  51  WHERE logs.object_uuid=containers.uuid
  52  AND logs.event_type in ('stdout', 'stderr', 'arv-mount', 'crunch-run', 'crunchstat', 'hoststat', 'node', 'container', 'keepstore')
  53  AND containers.log IS NOT NULL
  54  AND now() - containers.finished_at > $1::interval`,
  55                         h.Cluster.Containers.Logging.MaxAge.String())
  56                 if err != nil {
  57                         return err
  58                 }
  59                 logger := ctxlog.FromContext(ctx)
  60                 rows, err := res.RowsAffected()
  61                 if err != nil {
  62                         logger.WithError(err).Warn("unexpected error from RowsAffected()")
  63                 } else {
  64                         logger.WithField("rows", rows).Info("deleted rows from logs table")
  65                 }
  66                 return nil
  67         })
  68 }