Merge pull request PelicanPlatform#1300 from haoming29/issue-1281

Tune up file transfer tests

cache/self_monitor.go

@@ -30,13 +30,15 @@ import (
 	"strings"
 	"time"
 
+	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
+	"golang.org/x/sync/errgroup"
+
 	"github.com/pelicanplatform/pelican/config"
 	"github.com/pelicanplatform/pelican/metrics"
 	"github.com/pelicanplatform/pelican/param"
 	"github.com/pelicanplatform/pelican/token"
-	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
-	"golang.org/x/sync/errgroup"
+	"github.com/pelicanplatform/pelican/token_scopes"
 )
 
 const (
@@ -174,7 +176,7 @@ func generateFileTestScitoken() (string, error) {
 	fTestTokenCfg.Lifetime = time.Minute
 	fTestTokenCfg.Issuer = issuerUrl
 	fTestTokenCfg.Subject = "cache"
-	fTestTokenCfg.Claims = map[string]string{"scope": "storage.read:/pelican/monitoring/selfTest"}
+	fTestTokenCfg.AddResourceScopes(token_scopes.NewResourceScope(token_scopes.Storage_Read, "/pelican/monitoring/selfTest"))
 	// For self-tests, the audience is the server itself
 	fTestTokenCfg.AddAudienceAny()
 

director/cache_monitor.go

@@ -24,21 +24,23 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"path"
 	"strings"
 	"time"
 
-	"github.com/pelicanplatform/pelican/config"
 	"github.com/pkg/errors"
-)
 
-const (
-	cacheMonitroingBasePath = "/pelican/monitoring"
-	testFileContent         = "This object was created by the Pelican director-test functionality. Test is issued at "
+	"github.com/pelicanplatform/pelican/config"
+	"github.com/pelicanplatform/pelican/server_utils"
 )
 
+// Run one object transfer to a cache from the director. Since director-based cache tests require a different
+// workflow than the origin tests. We can't reuse server_utils.RunTests(), but we want to keep common
+// pieces together
 func runCacheTest(ctx context.Context, cacheUrl url.URL) error {
 	nowStr := time.Now().Format(time.RFC3339)
-	cacheUrl.Path = fmt.Sprintf("/pelican/monitoring/%s.txt", nowStr)
+	dirMonPath := path.Join(server_utils.MonitoringBaseNs, "directorTest")
+	cacheUrl = *cacheUrl.JoinPath(path.Join(dirMonPath, server_utils.DirectorTest.String()+"-"+nowStr+".txt"))
 	client := http.Client{Transport: config.GetTransport()}
 	req, err := http.NewRequestWithContext(ctx, "GET", cacheUrl.String(), nil)
 	if err != nil {
@@ -62,9 +64,9 @@ func runCacheTest(ctx context.Context, cacheUrl url.URL) error {
 	}
 	strBody := string(byteBody)
 
-	if strings.TrimSuffix(strBody, "\n") == testFileContent+nowStr {
+	if strings.TrimSuffix(strBody, "\n") == server_utils.DirectorTestBody {
 		return nil
 	} else {
-		return fmt.Errorf("cache response file does not match expectation. Expected:%s, Got:%s", testFileContent+nowStr, strBody)
+		return fmt.Errorf("cache response file does not match expectation. Expected:%s, Got:%s", server_utils.DirectorTestBody, strBody)
 	}
 }

director/director.go

@@ -41,6 +41,7 @@ import (
 	"github.com/pelicanplatform/pelican/metrics"
 	"github.com/pelicanplatform/pelican/param"
 	"github.com/pelicanplatform/pelican/server_structs"
+	"github.com/pelicanplatform/pelican/server_utils"
 	"github.com/pelicanplatform/pelican/token"
 	"github.com/pelicanplatform/pelican/token_scopes"
 )
@@ -953,16 +954,16 @@ func getPrefixByPath(ctx *gin.Context) {
 // Generate a mock file for caches to fetch. This is for director-based health tests for caches
 // So that we don't require an origin to feed the test file to the cache
 func getHealthTestFile(ctx *gin.Context) {
-	// Expected path: /pelican/monitoring/2006-01-02T15:04:05Z07:00.txt
+	// Expected path: /pelican/monitoring/directorTest/2006-01-02T15:04:05Z07:00.txt
 	pathParam := ctx.Param("path")
 	cleanedPath := path.Clean(pathParam)
-	if cleanedPath == "" || !strings.HasPrefix(cleanedPath, cacheMonitroingBasePath+"/") {
+	if cleanedPath == "" || !strings.HasPrefix(cleanedPath, server_utils.MonitoringBaseNs+"/") {
 		ctx.JSON(http.StatusBadRequest, server_structs.SimpleApiResp{
 			Status: server_structs.RespFailed,
 			Msg:    "Path parameter is not a valid health test path: " + cleanedPath})
 		return
 	}
-	fileName := strings.TrimPrefix(cleanedPath, cacheMonitroingBasePath+"/")
+	fileName := strings.TrimPrefix(cleanedPath, server_utils.MonitoringBaseNs+"/")
 	if fileName == "" {
 		ctx.JSON(http.StatusBadRequest, server_structs.SimpleApiResp{
 			Status: server_structs.RespFailed,
@@ -977,9 +978,7 @@ func getHealthTestFile(ctx *gin.Context) {
 		return
 	}
 
-	filenameWoExt := fileNameSplit[0]
-
-	fileContent := fmt.Sprintf("%s%s\n", testFileContent, filenameWoExt)
+	fileContent := server_utils.DirectorTestBody + "\n"
 
 	if ctx.Request.Method == "HEAD" {
 		ctx.Header("Content-Length", strconv.Itoa(len(fileContent)))

director/director_test.go

@@ -46,6 +46,7 @@ import (
 
 	"github.com/pelicanplatform/pelican/config"
 	"github.com/pelicanplatform/pelican/server_structs"
+	"github.com/pelicanplatform/pelican/server_utils"
 	"github.com/pelicanplatform/pelican/test_utils"
 	"github.com/pelicanplatform/pelican/token"
 	"github.com/pelicanplatform/pelican/token_scopes"
@@ -1203,7 +1204,7 @@ func TestGetHealthTestFile(t *testing.T) {
 
 		bytes, err := io.ReadAll(w.Result().Body)
 		require.NoError(t, err)
-		assert.Equal(t, testFileContent+"testfile\n", string(bytes))
+		assert.Equal(t, server_utils.DirectorTestBody+"\n", string(bytes))
 	})
 }
 

director/monitor.go

@@ -180,7 +180,7 @@ func LaunchPeriodicDirectorTest(ctx context.Context, serverAd server_structs.Ser
 			var err error
 			if serverAd.Type == server_structs.OriginType {
 				fileTests := server_utils.TestFileTransferImpl{}
-				ok, err = fileTests.RunTests(ctx, serverUrl, serverUrl, "", server_utils.DirectorFileTest)
+				ok, err = fileTests.RunTests(ctx, serverUrl, serverUrl, "", server_utils.DirectorTest)
 			} else if serverAd.Type == server_structs.CacheType {
 				err = runCacheTest(ctx, serverAd.URL)
 			}

origin/self_monitor.go

@@ -34,7 +34,7 @@ func doSelfMonitor(ctx context.Context) {
 	log.Debug("Starting a new self-test monitoring cycle")
 	fileTests := server_utils.TestFileTransferImpl{}
 	issuerUrl := param.Server_ExternalWebUrl.GetString()
-	ok, err := fileTests.RunTests(ctx, param.Origin_Url.GetString(), config.GetServerAudience(), issuerUrl, server_utils.OriginSelfFileTest)
+	ok, err := fileTests.RunTests(ctx, param.Origin_Url.GetString(), config.GetServerAudience(), issuerUrl, server_utils.ServerSelfTest)
 	if ok && err == nil {
 		log.Debugln("Self-test monitoring cycle succeeded at", time.Now().Format(time.UnixDate))
 		metrics.SetComponentHealthStatus(metrics.OriginCache_XRootD, metrics.StatusOK, "Self-test monitoring cycle succeeded at "+time.Now().Format(time.RFC3339))

server_utils/test_file_transfer.go

@@ -27,12 +27,14 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"path"
 	"time"
 
+	"github.com/pkg/errors"
+
 	"github.com/pelicanplatform/pelican/config"
 	"github.com/pelicanplatform/pelican/param"
 	"github.com/pelicanplatform/pelican/token"
-	"github.com/pkg/errors"
 )
 
 type (
@@ -45,22 +47,24 @@ type (
 		RunTests(ctx context.Context, baseUrl string, testType TestType) (bool, error)
 	}
 	TestFileTransferImpl struct {
-		audiences []string
-		issuerUrl string
-		testType  TestType
-		testBody  string
+		audiences    []string
+		issuerUrl    string
+		testType     TestType
+		testBody     string
+		testFilePath string // the path to the test file folder. e.g. /pelican/monitoring/selfTest
 	}
 )
 
 const (
-	OriginSelfFileTest TestType = "self-test"
-	DirectorFileTest   TestType = "director-test"
-	CacheTest          TestType = "cache-test"
+	ServerSelfTest TestType = "self-test"     // Origin/Cache object transfer self-test
+	DirectorTest   TestType = "director-test" // Director-based object transfer test
 )
 
+const MonitoringBaseNs string = "/pelican/monitoring" // The base namespace for monitoring objects
+
 const (
-	selfTestBody     string = "This object was created by the Pelican self-test functionality"
-	directorTestBody string = "This object was created by the Pelican director-test functionality"
+	SelfTestBody     string = "This object was created by the Pelican self-test functionality"
+	DirectorTestBody string = "This object was created by the Pelican director-test functionality"
 )
 
 func (t TestType) String() string {
@@ -89,13 +93,12 @@ func (t TestFileTransferImpl) generateFileTestScitoken() (string, error) {
 	if err != nil {
 		return "", errors.Wrap(err, "failed to create file test token")
 	}
-
 	return tok, nil
 }
 
 // Private function to upload a test file to the `baseUrl` of an exported xrootd file directory
 // the test file content is based on the `testType` attribute
-func (t TestFileTransferImpl) uploadTestfile(ctx context.Context, baseUrl string, namespace string) (string, error) {
+func (t TestFileTransferImpl) uploadTestfile(ctx context.Context, baseUrl string) (string, error) {
 	tkn, err := t.generateFileTestScitoken()
 	if err != nil {
 		return "", errors.Wrap(err, "Failed to create a token for test file transfer")
@@ -105,7 +108,8 @@ func (t TestFileTransferImpl) uploadTestfile(ctx context.Context, baseUrl string
 	if err != nil {
 		return "", errors.Wrap(err, "The baseUrl is not parseable as a URL")
 	}
-	uploadURL.Path = namespace + t.testType.String() + "-" + time.Now().Format(time.RFC3339) + ".txt"
+	// /pelican/monitoring/<selfTest|directorTest>/<self-test|director-test>-YYYY-MM-DDTHH:MM:SSZ.txt
+	uploadURL = uploadURL.JoinPath(path.Join(t.testFilePath, t.testType.String()+"-"+time.Now().Format(time.RFC3339)+".txt"))
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodPut, uploadURL.String(), bytes.NewBuffer([]byte(t.testBody)))
 	if err != nil {
@@ -208,15 +212,19 @@ func (t TestFileTransferImpl) RunTests(ctx context.Context, baseUrl, audienceUrl
 	t.audiences = []string{baseUrl, audienceUrl}
 	t.issuerUrl = issuerUrl
 	t.testType = testType
-	if testType == OriginSelfFileTest {
-		t.testBody = selfTestBody
-	} else if testType == DirectorFileTest {
-		t.testBody = directorTestBody
+	t.testFilePath = MonitoringBaseNs
+
+	if t.testType == ServerSelfTest {
+		t.testBody = SelfTestBody
+		t.testFilePath = path.Join(MonitoringBaseNs, "selfTest")
+	} else if t.testType == DirectorTest {
+		t.testBody = DirectorTestBody
+		t.testFilePath = path.Join(MonitoringBaseNs, "directorTest")
 	} else {
 		return false, errors.New("Unsupported testType: " + testType.String())
 	}
 
-	downloadUrl, err := t.uploadTestfile(ctx, baseUrl, "/pelican/monitoring/")
+	downloadUrl, err := t.uploadTestfile(ctx, baseUrl)
 	if err != nil {
 		return false, errors.Wrap(err, "Test file transfer failed during upload")
 	}

xrootd/origin_test.go

@@ -180,7 +180,7 @@ func TestOrigin(t *testing.T) {
 	issuerUrl, err := config.GetServerIssuerURL()
 	require.NoError(t, err)
 
-	ok, err := fileTests.RunTests(ctx, param.Origin_Url.GetString(), config.GetServerAudience(), issuerUrl, server_utils.OriginSelfFileTest)
+	ok, err := fileTests.RunTests(ctx, param.Origin_Url.GetString(), config.GetServerAudience(), issuerUrl, server_utils.ServerSelfTest)
 	require.NoError(t, err)
 	require.True(t, ok)
 }
@@ -228,7 +228,7 @@ func TestMultiExportOrigin(t *testing.T) {
 	issuerUrl, err := config.GetServerIssuerURL()
 	require.NoError(t, err)
 
-	ok, err := fileTests.RunTests(ctx, param.Origin_Url.GetString(), config.GetServerAudience(), issuerUrl, server_utils.OriginSelfFileTest)
+	ok, err := fileTests.RunTests(ctx, param.Origin_Url.GetString(), config.GetServerAudience(), issuerUrl, server_utils.ServerSelfTest)
 	require.NoError(t, err)
 	require.True(t, ok)
 }