Allow configurable cache watermarks

config/config.go

@@ -31,6 +31,7 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -143,6 +144,8 @@ var (
 	version string = "dev"
 
 	MetadataTimeoutErr *MetadataErr = &MetadataErr{msg: "Timeout when querying metadata"}
+
+	watermarkUnits = []byte{'k', 'm', 'g', 't'}
 )
 
 // This function creates a new MetadataError by wrapping the previous error
@@ -659,6 +662,33 @@ func handleDeprecatedConfig() {
 	}
 }
 
+func checkWatermark(wmStr string) (bool, error) {
+	wmNum, err := strconv.ParseFloat(wmStr, 64)
+	// Not a float number, check if it's in form of <int>k|m|g|t
+	if err != nil {
+		if len(wmStr) < 1 {
+			return false, errors.Errorf("watermark value %s is empty.", wmStr)
+		}
+		if slices.Contains(watermarkUnits, wmStr[len(wmStr)-1]) {
+			_, err := strconv.Atoi(wmStr[:len(wmStr)-1])
+			// Bytes portion is not an integer
+			if err != nil {
+				return false, errors.Errorf("watermark value %s is neither a decimal fraction of a percentage number (e.g. 0.95) or a valid bytes. Refer to parameter page for details: https://docs.pelicanplatform.org/parameters#Cache-HighWatermark", wmStr)
+			} else {
+				return true, nil
+			}
+		} else {
+			// Doesn't contain k|m|g|t suffix
+			return false, errors.Errorf("watermark value %s is neither a decimal fraction of a percentage number (e.g. 0.95) or a valid bytes. Bytes representation is missing unit (k|m|g|t). Refer to parameter page for details: https://docs.pelicanplatform.org/parameters#Cache-HighWatermark", wmStr)
+		}
+	} else {
+		if wmNum > 1 || wmNum < 0 {
+			return false, errors.Errorf("watermark value %s must be a float number in range [0.0, 1.0]. Refer to parameter page for details: https://docs.pelicanplatform.org/parameters#Cache-HighWatermark", wmStr)
+		}
+		return true, nil
+	}
+}
+
 func InitConfig() {
 	viper.SetConfigType("yaml")
 	// 1) Set up defaults.yaml
@@ -986,6 +1016,19 @@ func InitServer(ctx context.Context, currentServers ServerType) error {
 		viper.SetDefault("Cache.Url", fmt.Sprintf("https://%v", param.Server_Hostname.GetString()))
 	}
 
+	if param.Cache_HighWaterMark.IsSet() && param.Cache_LowWatermark.IsSet() {
+		highWmStr := param.Cache_HighWaterMark.GetString()
+		lowWmStr := param.Cache_LowWatermark.GetString()
+		ok, err := checkWatermark(highWmStr)
+		if !ok && err != nil {
+			return errors.Wrap(err, "invalid Cache.HighWaterMark value")
+		}
+		ok, err = checkWatermark(lowWmStr)
+		if !ok && err != nil {
+			return errors.Wrap(err, "invalid Cache.LowWatermark value")
+		}
+	}
+
 	webPort := param.Server_WebPort.GetInt()
 	if webPort < 0 {
 		return errors.Errorf("the Server.WebPort setting of %d is invalid; TCP ports must be greater than 0", webPort)

config/config_test.go

@@ -342,3 +342,91 @@ func TestDiscoverFederation(t *testing.T) {
 		viper.Reset()
 	})
 }
+
+func TestCheckWatermark(t *testing.T) {
+	t.Parallel()
+
+	t.Run("empty-value", func(t *testing.T) {
+		ok, err := checkWatermark("")
+		assert.False(t, ok)
+		assert.Error(t, err)
+	})
+
+	t.Run("string-value", func(t *testing.T) {
+		ok, err := checkWatermark("random")
+		assert.False(t, ok)
+		assert.Error(t, err)
+	})
+
+	t.Run("integer-greater-than-1", func(t *testing.T) {
+		ok, err := checkWatermark("2")
+		assert.False(t, ok)
+		assert.Error(t, err)
+	})
+
+	t.Run("integer-less-than-0", func(t *testing.T) {
+		ok, err := checkWatermark("-1")
+		assert.False(t, ok)
+		assert.Error(t, err)
+	})
+
+	t.Run("float-value-single-decimal", func(t *testing.T) {
+		ok, err := checkWatermark("0.5")
+		assert.True(t, ok)
+		assert.NoError(t, err)
+	})
+
+	t.Run("float-value-two-decimals", func(t *testing.T) {
+		ok, err := checkWatermark("0.55")
+		assert.True(t, ok)
+		assert.NoError(t, err)
+	})
+
+	t.Run("byte-value-no-unit", func(t *testing.T) {
+		ok, err := checkWatermark("100")
+		assert.False(t, ok)
+		assert.Error(t, err)
+	})
+
+	t.Run("byte-value-no-value", func(t *testing.T) {
+		ok, err := checkWatermark("k")
+		assert.False(t, ok)
+		assert.Error(t, err)
+	})
+
+	t.Run("byte-value-wrong-unit", func(t *testing.T) {
+		ok, err := checkWatermark("100K") // Only lower case is accepted
+		assert.False(t, ok)
+		assert.Error(t, err)
+
+		ok, err = checkWatermark("100p")
+		assert.False(t, ok)
+		assert.Error(t, err)
+
+		ok, err = checkWatermark("100byte")
+		assert.False(t, ok)
+		assert.Error(t, err)
+
+		ok, err = checkWatermark("100bits")
+		assert.False(t, ok)
+		assert.Error(t, err)
+	})
+
+	t.Run("byte-value-correct-unit", func(t *testing.T) {
+		ok, err := checkWatermark("1000k")
+		assert.True(t, ok)
+		assert.NoError(t, err)
+
+		ok, err = checkWatermark("1000m")
+		assert.True(t, ok)
+		assert.NoError(t, err)
+
+		ok, err = checkWatermark("1000g")
+		assert.True(t, ok)
+		assert.NoError(t, err)
+
+		ok, err = checkWatermark("1000t")
+		assert.True(t, ok)
+		assert.NoError(t, err)
+	})
+}

config/resources/defaults.yaml

@@ -49,6 +49,8 @@ Cache:
   Port: 8442
   SelfTest: true
   SelfTestInterval: 15s
+  LowWatermark: 0.90
+  HighWaterMark: 0.95
 LocalCache:
   HighWaterMarkPercentage: 95
   LowWaterMarkPercentage: 85

docs/parameters.yaml

@@ -857,6 +857,28 @@ type: int
 default: 8442
 components: ["cache"]
 ---
+name: Cache.LowWatermark
+description: >-
+  A value of cache disk usage that stops the purging of cached files.
+
+  The value should be either a decimal fraction of a percentage number of total avaibale disk space (default is 0.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.
+type: string
+default: 0.90
+components: ["cache"]
+---
+name: Cache.HighWaterMark
+description: >-
+  A value of cache disk usage that triggers the purging of cached files.
+
+  The value should be either a decimal fraction of a percentage number of total avaibale disk space (default is 0.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.
+type: string
+default: 0.95
+components: ["cache"]
+---
 name: Cache.EnableVoms
 description: >-
   Enable X.509 / VOMS-based authentication for the cache.  This allows HTTP clients

xrootd/resources/xrootd-cache.cfg

@@ -55,7 +55,8 @@ pfc.blocksize 128k
 pfc.prefetch 20
 pfc.writequeue 16 4
 pfc.ram 4g
-pfc.diskusage 0.90 0.95 purgeinterval 300s
+pfc.diskusage {{if .Cache.LowWatermark}}{{.Cache.LowWatermark}}{{else}}0.90{{end}} {{if .Cache.HighWaterMark}}{{.Cache.HighWaterMark}}{{else}}0.95{{end}} purgeinterval 300s
+
 {{if .Cache.Concurrency}}
 xrootd.fslib throttle default
 throttle.throttle concurrency {{.Cache.Concurrency}}

xrootd/xrootd_config.go

@@ -104,6 +104,8 @@ type (
 		UseCmsd        bool
 		EnableVoms     bool
 		CalculatedPort string
+		HighWaterMark  string
+		LowWatermark   string
 		ExportLocation string
 		RunLocation    string
 		DataLocation   string