Merge pull request #973 from haoming29/cache-watermarks

Allow configurable cache watermarks

config/config.go

@@ -31,6 +31,7 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -158,7 +159,8 @@ var (
 
 	MetadataTimeoutErr *MetadataErr = &MetadataErr{msg: "Timeout when querying metadata"}
 
-	validPrefixes = map[string]bool{
+	watermarkUnits = []byte{'k', 'm', 'g', 't'}
+	validPrefixes  = map[string]bool{
 		string(Pelican): true,
 		string(OSDF):    true,
 		string(Stash):   true,
@@ -742,6 +744,45 @@ 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 InitConfig() {
 	viper.SetConfigType("yaml")
 	// 1) Set up defaults.yaml
@@ -1099,6 +1140,22 @@ func InitServer(ctx context.Context, currentServers ServerType) error {
 		viper.SetDefault("Cache.Url", fmt.Sprintf("https://%v", param.Server_Hostname.GetString()))
 	}
 
+	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")
+		}
+		ok, lowWmNum, err := checkWatermark(lowWmStr)
+		if !ok && err != nil {
+			return errors.Wrap(err, "invalid Cache.LowWatermark value")
+		}
+		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)
+		}
+	}
+
 	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

@@ -384,6 +384,117 @@ func TestDiscoverFederation(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) {
 	mockHostname := "example.com"
 	mockNon443Port := 8444

config/resources/defaults.yaml

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

docs/parameters.yaml

@@ -872,6 +872,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 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.
+type: string
+default: 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 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.
+type: string
+default: 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

@@ -53,7 +53,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

@@ -34,6 +34,7 @@ import (
 	"path/filepath"
 	"reflect"
 	"runtime"
+	"strconv"
 	"strings"
 	"text/template"
 	"time"
@@ -104,6 +105,8 @@ type (
 		UseCmsd        bool
 		EnableVoms     bool
 		CalculatedPort string
+		HighWaterMark  string
+		LowWatermark   string
 		ExportLocation string
 		RunLocation    string
 		DataLocation   string
@@ -591,6 +594,20 @@ func ConfigXrootd(ctx context.Context, origin bool) (string, error) {
 		return "", errors.Wrap(err, "failed to unmarshal xrootd config")
 	}
 
+	// For cache. convert integer percentage value [0,100] to decimal fraction [0.00, 1.00]
+	if !origin {
+		if num, err := strconv.Atoi(xrdConfig.Cache.HighWaterMark); err == nil {
+			if num <= 100 && num > 0 {
+				xrdConfig.Cache.HighWaterMark = strconv.FormatFloat(float64(num)/100, 'f', 2, 64)
+			}
+		}
+		if num, err := strconv.Atoi(xrdConfig.Cache.LowWatermark); err == nil {
+			if num <= 100 && num > 0 {
+				xrdConfig.Cache.LowWatermark = strconv.FormatFloat(float64(num)/100, 'f', 2, 64)
+			}
+		}
+	}
+
 	// To make sure we get the correct exports, we overwrite the exports in the xrdConfig struct with the exports
 	// we get from the server_structs.GetOriginExports() function. Failure to do so will cause us to hit viper again,
 	// which in the case of tests prevents us from overwriting some exports with temp dirs.