Merge pull request #1935 from jhiemstrawisc/lotman-plumbing

Plumb Lotman config through to XRootD purge plugin

Author: Saartank

config/config.go

@@ -33,7 +33,6 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
-	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -55,6 +54,7 @@ import (
 	"github.com/pelicanplatform/pelican/param"
 	"github.com/pelicanplatform/pelican/pelican_url"
 	"github.com/pelicanplatform/pelican/server_structs"
+	"github.com/pelicanplatform/pelican/utils"
 )
 
 // Structs holding the OAuth2 state (and any other OSDF config needed)
@@ -165,8 +165,7 @@ var (
 
 	RestartFlag = make(chan any) // A channel flag to restart the server instance that launcher listens to (including cache)
 
-	watermarkUnits = []byte{'k', 'm', 'g', 't'}
-	validPrefixes  = map[ConfigPrefix]bool{
+	validPrefixes = map[ConfigPrefix]bool{
 		PelicanPrefix: true,
 		OsdfPrefix:    true,
 		StashPrefix:   true,
@@ -627,45 +626,6 @@ func handleDeprecatedConfig() {
 	}
 }
 
-func checkWatermark(wmStr string) (bool, int64, error) {
-	wmNum, err := strconv.Atoi(wmStr)
-	if err == nil {
-		if wmNum > 100 || wmNum < 0 {
-			return false, 0, errors.Errorf("watermark value %s must be a integer number in range [0, 100]. Refer to parameter page for details: https://docs.pelicanplatform.org/parameters#Cache-HighWatermark", wmStr)
-		}
-		return true, int64(wmNum), nil
-		// Not an integer number, check if it's in form of <int>k|m|g|t
-	} else {
-		if len(wmStr) < 1 {
-			return false, 0, errors.Errorf("watermark value %s is empty.", wmStr)
-		}
-		unit := wmStr[len(wmStr)-1]
-		if slices.Contains(watermarkUnits, unit) {
-			byteNum, err := strconv.Atoi(wmStr[:len(wmStr)-1])
-			// Bytes portion is not an integer
-			if err != nil {
-				return false, 0, errors.Errorf("watermark value %s is neither a percentage integer (e.g. 95) or a valid bytes. Refer to parameter page for details: https://docs.pelicanplatform.org/parameters#Cache-HighWatermark", wmStr)
-			} else {
-				switch unit {
-				case 'k':
-					return true, int64(byteNum) * 1024, nil
-				case 'm':
-					return true, int64(byteNum) * 1024 * 1024, nil
-				case 'g':
-					return true, int64(byteNum) * 1024 * 1024 * 1024, nil
-				case 't':
-					return true, int64(byteNum) * 1024 * 1024 * 1024 * 1024, nil
-				default:
-					return false, 0, errors.Errorf("watermark value %s is neither a percentage integer (e.g. 95) or a valid byte. Bytes representation is missing unit (k|m|g|t). Refer to parameter page for details: https://docs.pelicanplatform.org/parameters#Cache-HighWatermark", wmStr)
-				}
-			}
-		} else {
-			// Doesn't contain k|m|g|t suffix
-			return false, 0, errors.Errorf("watermark value %s is neither a percentage integer (e.g. 95) or a valid byte. Bytes representation is missing unit (k|m|g|t). Refer to parameter page for details: https://docs.pelicanplatform.org/parameters#Cache-HighWatermark", wmStr)
-		}
-	}
-}
-
 func setupTranslation() error {
 	err := en_translations.RegisterDefaultTranslations(validate, GetEnTranslator())
 	if err != nil {
@@ -1331,18 +1291,63 @@ func InitServer(ctx context.Context, currentServers server_structs.ServerType) e
 	}
 
 	if param.Cache_LowWatermark.IsSet() || param.Cache_HighWaterMark.IsSet() {
-		lowWmStr := param.Cache_LowWatermark.GetString()
-		highWmStr := param.Cache_HighWaterMark.GetString()
-		ok, highWmNum, err := checkWatermark(highWmStr)
-		if !ok && err != nil {
-			return errors.Wrap(err, "invalid Cache.HighWaterMark value")
+		lowWm, lwmIsAbs, err := utils.ValidateWatermark(param.Cache_LowWatermark.GetName(), false)
+		if err != nil {
+			return err
 		}
-		ok, lowWmNum, err := checkWatermark(lowWmStr)
-		if !ok && err != nil {
-			return errors.Wrap(err, "invalid Cache.LowWatermark value")
+		highWm, hwmIsAbs, err := utils.ValidateWatermark(param.Cache_HighWaterMark.GetName(), false)
+		if err != nil {
+			return err
+		}
+		if lwmIsAbs == hwmIsAbs && lowWm >= highWm {
+			// Use config strings in error to present the configured values (as opposed to whatever conversion we get from validation)
+			return errors.Errorf("invalid Cache.HighWaterMark/LowWaterMark values. The high watermark must be greater than the low "+
+				"watermark. Got %s (low) and %s (high)", param.Cache_LowWatermark.GetString(), param.Cache_HighWaterMark.GetString())
+		}
+	}
+
+	if param.Cache_FilesBaseSize.IsSet() || param.Cache_FilesNominalSize.IsSet() || param.Cache_FilesMaxSize.IsSet() {
+		// Must have high/low watermarks
+		if !param.Cache_LowWatermark.IsSet() || !param.Cache_HighWaterMark.IsSet() {
+			return errors.New("If any of Cache parameters 'FilesBaseSize', 'FilesNominalSize', or 'FilesMaxSize' is set, then Cache.LowWatermark " +
+				"and Cache.HighWatermark must also be set")
+		}
+
+		// Further, if one is set, all three must be set
+		if !param.Cache_FilesBaseSize.IsSet() || !param.Cache_FilesNominalSize.IsSet() || !param.Cache_FilesMaxSize.IsSet() {
+			return errors.New("If any of Cache parameters 'FilesBaseSize', 'FilesNominalSize', or 'FilesMaxSize' is set, all three must be set")
 		}
-		if lowWmNum >= highWmNum {
-			return fmt.Errorf("invalid Cache.HighWaterMark and Cache.LowWatermark values. Cache.HighWaterMark must be greater than Cache.LowWaterMark. Got %s, %s", highWmStr, lowWmStr)
+
+		var base, nominal, max float64
+		var err error
+		// Watermark validation will error if these parameters are not absolute, so we can ignore that output
+		if base, _, err = utils.ValidateWatermark(param.Cache_FilesBaseSize.GetName(), true); err != nil {
+			return err
+		}
+		if nominal, _, err = utils.ValidateWatermark(param.Cache_FilesNominalSize.GetName(), true); err != nil {
+			return err
+		}
+		if max, _, err = utils.ValidateWatermark(param.Cache_FilesMaxSize.GetName(), true); err != nil {
+			return err
+		}
+		if base >= nominal || nominal >= max {
+			return errors.Errorf("invalid Cache.FilesBaseSize/FilesNominalSize/FilesMaxSize values. The base size must be less than the "+
+				"nominal size, and the nominal size must be less than the max size. Got %s (base), %s (nominal), and %s (max)",
+				param.Cache_FilesBaseSize.GetString(), param.Cache_FilesNominalSize.GetString(), param.Cache_FilesMaxSize.GetString())
+		}
+
+		// File sizes must also be less than the low watermark, but that's not straightforward to check, especially when the cache spread across
+		// multiple disks and the low watermark is configured as a relative value. If such bad config is passed, xrootd will fail to startup with
+		// a message about what to do, and as much as I'd like to handle that early, I'll leave it to xrootd for now.
+	}
+	if param.Cache_EnableLotman.GetBool() {
+		// pfc.diskusage file directives _must_ be set.
+		// We already validate that one means all three, but I'm duplicating that here to make this safer in the case we switch orders
+		// in the future.
+		if !param.Cache_FilesBaseSize.IsSet() || !param.Cache_FilesNominalSize.IsSet() || !param.Cache_FilesMaxSize.IsSet() ||
+			!param.Cache_LowWatermark.IsSet() || !param.Cache_HighWaterMark.IsSet() {
+			return errors.New("If Cache.EnableLotman is true, the following Cache parameters must also be set: HighWaterMark, LowWaterMark, " +
+				"FilesBaseSize, FilesNominalSize, FilesMaxSize")
 		}
 	}
 

config/config_test.go

@@ -549,117 +549,6 @@ func TestSetPreferredPrefix(t *testing.T) {
 	})
 }
 
-func TestCheckWatermark(t *testing.T) {
-	t.Parallel()
-
-	t.Run("empty-value", func(t *testing.T) {
-		ok, num, err := checkWatermark("")
-		assert.False(t, ok)
-		assert.Equal(t, int64(0), num)
-		assert.Error(t, err)
-	})
-
-	t.Run("string-value", func(t *testing.T) {
-		ok, num, err := checkWatermark("random")
-		assert.False(t, ok)
-		assert.Equal(t, int64(0), num)
-		assert.Error(t, err)
-	})
-
-	t.Run("integer-greater-than-100", func(t *testing.T) {
-		ok, num, err := checkWatermark("101")
-		assert.False(t, ok)
-		assert.Equal(t, int64(0), num)
-		assert.Error(t, err)
-	})
-
-	t.Run("integer-less-than-0", func(t *testing.T) {
-		ok, num, err := checkWatermark("-1")
-		assert.False(t, ok)
-		assert.Equal(t, int64(0), num)
-		assert.Error(t, err)
-	})
-
-	t.Run("decimal-fraction-value", func(t *testing.T) {
-		ok, num, err := checkWatermark("0.55")
-		assert.False(t, ok)
-		assert.Equal(t, int64(0), num)
-		assert.Error(t, err)
-	})
-
-	t.Run("decimal-int-value", func(t *testing.T) {
-		ok, num, err := checkWatermark("15.55")
-		assert.False(t, ok)
-		assert.Equal(t, int64(0), num)
-		assert.Error(t, err)
-	})
-
-	t.Run("int-value", func(t *testing.T) {
-		ok, num, err := checkWatermark("55")
-		assert.True(t, ok)
-		assert.Equal(t, int64(55), num)
-		assert.NoError(t, err)
-	})
-
-	t.Run("byte-value-no-unit", func(t *testing.T) {
-		ok, num, err := checkWatermark("105")
-		assert.False(t, ok)
-		assert.Equal(t, int64(0), num)
-		assert.Error(t, err)
-	})
-
-	t.Run("byte-value-no-value", func(t *testing.T) {
-		ok, num, err := checkWatermark("k")
-		assert.False(t, ok)
-		assert.Equal(t, int64(0), num)
-		assert.Error(t, err)
-	})
-
-	t.Run("byte-value-wrong-unit", func(t *testing.T) {
-		ok, num, err := checkWatermark("100K") // Only lower case is accepted
-		assert.False(t, ok)
-		assert.Equal(t, int64(0), num)
-		assert.Error(t, err)
-
-		ok, num, err = checkWatermark("100p")
-		assert.False(t, ok)
-		assert.Equal(t, int64(0), num)
-		assert.Error(t, err)
-
-		ok, num, err = checkWatermark("100byte")
-		assert.False(t, ok)
-		assert.Equal(t, int64(0), num)
-		assert.Error(t, err)
-
-		ok, num, err = checkWatermark("100bits")
-		assert.False(t, ok)
-		assert.Equal(t, int64(0), num)
-		assert.Error(t, err)
-	})
-
-	t.Run("byte-value-correct-unit", func(t *testing.T) {
-		ok, num, err := checkWatermark("1000k")
-		assert.True(t, ok)
-		assert.Equal(t, int64(1000*1024), num)
-		assert.NoError(t, err)
-
-		ok, num, err = checkWatermark("1000m")
-		assert.True(t, ok)
-		assert.Equal(t, int64(1000*1024*1024), num)
-		assert.NoError(t, err)
-
-		ok, num, err = checkWatermark("1000g")
-		assert.True(t, ok)
-		assert.Equal(t, int64(1000*1024*1024*1024), num)
-		assert.NoError(t, err)
-
-		ok, num, err = checkWatermark("1000t")
-		assert.True(t, ok)
-		assert.Equal(t, int64(1000*1024*1024*1024*1024), num)
-		assert.NoError(t, err)
-	})
-}
-
 func TestInitServerUrl(t *testing.T) {
 	ctx := testConfigContext(t)
 

docs/parameters.yaml

@@ -1326,26 +1326,83 @@ components: ["cache"]
 ---
 name: Cache.LowWatermark
 description: |+
-  A value of cache disk usage that stops the purging of cached files.
+  Whenever the cache initiates file purging, it will attempt to clean files until its cumulative disk
+  usages reaches this value. Note that "cache disk usage" is calculated based on the cache's entire set of
+  configured disks, not just data directories from those disks.
 
   The value should be either a percentage integer of total available disk space (default is 90),
   or a number suffixed by k, m, g, or t. In which case, they must be absolute sizes in k (kilo-),
   m (mega-), g (giga-), or t (tera-) bytes, respectively.
+
+  For more information, see the [xrootd pfc documentation](https://xrootd.web.cern.ch/doc/dev56/pss_config.pdf) for
+  `pfc.diskusage`.
 type: string
 default: 90
 components: ["cache"]
 ---
 name: Cache.HighWaterMark
 description: |+
-  A value of cache disk usage that triggers the purging of cached files.
+  When the cache's disk usage exceeds this value, file purging is triggered. Note that "cache disk usage"
+  is calculated based on the cache's entire set of configured disks, not just data directories from those disks.
 
   The value should be either a percentage integer of total available disk space (default is 95),
   or a number suffixed by k, m, g, or t. In which case, they must be absolute sizes in k (kilo-),
   m (mega-), g (giga-), or t (tera-) bytes, respectively.
+
+  For more information, see the [xrootd pfc documentation](https://xrootd.web.cern.ch/doc/dev56/pss_config.pdf) for
+  `pfc.diskusage`.
 type: string
 default: 95
 components: ["cache"]
 ---
+name: Cache.FilesMaxSize
+description: |+
+  A value that sets the maximum cumulative size of files that can be stored in the cache's data directories (specified
+  by `Cache.StorageLocation` and `Cache.DataLocations). When either this value or `Cache.HighWaterMark` is exceeded, the
+  cache will begin purging files until it reaches the `Cache.FilesNominal` value. If the cache's disk usage is still in
+  excess of the `Cache.LowWaterMark`, the cache will continue purging files until it reaches the `Cache.FilesBase` value.
+
+  Unlike watermark values, this value _must_ be suffixed by a unit of k, m, g, or t, which represent kilobytes, megabytes,
+  gigabytes, and terabytes, respectively. All `Cache.Files*Size` parameters must be less than the cache's calculated low
+  watermark, which may be configured as a percentage of total disk space from multiple disks.
+
+  For more information, see the [xrootd pfc documentation](https://xrootd.web.cern.ch/doc/dev56/pss_config.pdf) for
+  `pfc.diskusage`.
+type: string
+default: none
+components: ["cache"]
+---
+name: Cache.FilesNominalSize
+description: |+
+  A value that sets the "nominal" cumulative size of files that can be stored in the cache's data directory. When files
+  in the cache exceed the `Cache.FilesMax` value, or if the cache's overall disk exceeds its `Cache.HighWaterMark` value,
+  the cache will begin purging files until it reaches this value. If the cache's disk usage is still in excess of the
+  `Cache.LowWaterMark`, the cache will continue purging files until it reaches the `Cache.FilesBase`.
+
+  Unlike watermark values, this value _must_ be suffixed by a unit of k, m, g, or t, which represent kilobytes, megabytes,
+  gigabytes, and terabytes, respectively. All `Cache.Files*Size` parameters must be less than the cache's calculated low
+  watermark, which may be configured as a percentage of total disk space from multiple disks.
+
+  For more information, see the [xrootd pfc documentation](https://xrootd.web.cern.ch/doc/dev56/pss_config.pdf) for
+  `pfc.diskusage`.
+type: string
+default: none
+components: ["cache"]
+---
+name: Cache.FilesBaseSize
+description: |+
+  A value that sets the "base" cumulative size of files that can be stored in the cache's data directory. This is the
+  stopping point for the cache's purging routines.
+
+  Unlike watermark values, this value _must_ be suffixed by a unit of k, m, g, or t, which represent kilobytes, megabytes,
+  gigabytes, and terabytes, respectively.All `Cache.Files*Size` parameters must be less than the cache's calculated low
+  watermark, which may be configured as a percentage of total disk space from multiple disks.
+
+  For more information, see the [xrootd pfc documentation](https://xrootd.web.cern.ch/doc/dev56/pss_config.pdf) for `pfc.diskusage`.
+type: string
+default: none
+components: ["cache"]
+---
 name: Cache.EnableVoms
 description: |+
   Enable X.509 / VOMS-based authentication for the cache.  This allows HTTP clients

lotman/lotman.go

@@ -33,6 +33,13 @@ import (
 	"github.com/pelicanplatform/pelican/server_structs"
 )
 
+// Minimal re-definition of PurgePolicy so that things compile on non-Linux platforms.
+// We may use a different approach someday so we don't need multiple definitions of this struct
+// but for now I think this is okay...
+type PurgePolicy struct {
+	PurgeOrder []string
+}
+
 func RegisterLotman(ctx context.Context, router *gin.RouterGroup) {
 	log.Warningln("LotMan is not supported on this platform. Skipping...")
 }
@@ -41,3 +48,8 @@ func InitLotman(adsFromFed []server_structs.NamespaceAdV2) bool {
 	log.Warningln("LotMan is not supported on this platform. Skipping...")
 	return false
 }
+
+func GetPolicyMap() (map[string]PurgePolicy, error) {
+	log.Warningln("LotMan is not supported on this platform. Skipping...")
+	return map[string]PurgePolicy{}, nil
+}