Provide an E2E cache test for offline origins

Shows that data can be accessed at a cache even when the origin
is offline!

Author: bbockelm

e2e_fed_tests/cache_scitokens_config_test.go

@@ -21,17 +21,22 @@
 package fed_tests
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/url"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
 	_ "github.com/glebarez/sqlite"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/pelicanplatform/pelican/cache"
 	"github.com/pelicanplatform/pelican/client"
 	"github.com/pelicanplatform/pelican/config"
 	"github.com/pelicanplatform/pelican/fed_test_utils"
@@ -41,17 +46,16 @@ import (
 	"github.com/pelicanplatform/pelican/test_utils"
 	"github.com/pelicanplatform/pelican/token"
 	"github.com/pelicanplatform/pelican/token_scopes"
-	"github.com/pelicanplatform/pelican/xrootd"
 )
 
 // TestCacheScitokensConfigOverride tests that Xrootd.ScitokensConfig works for caches
 // to serve cached objects during origin downtime. This test:
 // 1. Sets up a full federation with private reads and pulls a file through the cache
-// 2. Simulates origin downtime by removing namespace ads from the cache
+// 2. Simulates origin downtime by POSTing a new origin ad without the /test namespace
 // 3. Triggers cache authz refresh by overwriting Xrootd.ScitokensConfig with unrelated issuer
-// 4. Verifies data is no longer accessible through the cache
+// 4. Verifies data is no longer accessible through the cache (authorization removed)
 // 5. Triggers another authz refresh with proper authorization for the test prefix
-// 6. Verifies cached object is now accessible even without origin
+// 6. Verifies cached object is now accessible even with origin "offline"
 func TestCacheScitokensConfigOverride(t *testing.T) {
 	t.Cleanup(test_utils.SetupTestLogging(t))
 	server_utils.ResetTestState()
@@ -75,6 +79,12 @@ func TestCacheScitokensConfigOverride(t *testing.T) {
 	scitokensConfigPath := filepath.Join(tmpDir, "scitokens.cfg")
 	require.NoError(t, param.Set(param.Xrootd_ScitokensConfig.GetName(), scitokensConfigPath))
 
+	// Set floor to 0 to allow immediate director refreshes when scitokens config changes
+	require.NoError(t, param.Set(param.Cache_MinDirectorRefreshInterval.GetName(), "0s"))
+
+	// Use long-lived ads so they don't expire during test
+	require.NoError(t, param.Set(param.Server_AdLifetime.GetName(), "1h"))
+
 	// Create origin configuration with private reads (requires tokens)
 	originConfig := `
 Origin:
@@ -92,38 +102,124 @@ Origin:
 	serverIssuerUrl, err := config.GetServerIssuerURL()
 	require.NoError(t, err, "Failed to get server issuer URL")
 
-	// Get federation info
-	fedInfo, err := config.GetFederation(ctx)
-	require.NoError(t, err)
-
 	// Create a token for accessing the object
 	tokenConfig := token.NewWLCGToken()
 	tokenConfig.Lifetime = 30 * time.Minute
 	tokenConfig.Issuer = serverIssuerUrl
 	tokenConfig.Subject = "test-subject"
-	tokenConfig.AddAudiences("https://wlcg.cern.ch/jwt/v1/any")
-	tokenConfig.AddResourceScopes(token_scopes.NewResourceScope(token_scopes.Wlcg_Storage_Read, "/test"))
+	tokenConfig.AddAudienceAny()
+
+	scopes := []token_scopes.TokenScope{}
+	readScope, err := token_scopes.Wlcg_Storage_Read.Path("/")
+	require.NoError(t, err)
+	scopes = append(scopes, readScope)
+	modScope, err := token_scopes.Wlcg_Storage_Modify.Path("/")
+	require.NoError(t, err)
+	scopes = append(scopes, modScope)
+	tokenConfig.AddScopes(scopes...)
 
 	tok, err := tokenConfig.CreateToken()
 	require.NoError(t, err)
 
+	// Construct pelican URL for file operations
+	pelicanUrl := fmt.Sprintf("pelican://%s:%d/test/%s",
+		param.Server_Hostname.GetString(), param.Server_WebPort.GetInt(), testFileName)
+
+	// Upload the test file to the origin
+	_, err = client.DoPut(ctx, testFilePath, pelicanUrl, false, client.WithToken(tok))
+	require.NoError(t, err, "Should be able to upload file to origin")
+
 	// Step 1: Download through the federation to populate the cache
 	destPath1 := filepath.Join(tmpDir, "downloaded1.txt")
-	transferResults, err := client.DoGet(ctx, "pelican://"+fedInfo.DirectorEndpoint+"/test/"+testFileName, destPath1, false, client.WithToken(tok))
+	transferResults, err := client.DoGet(ctx, pelicanUrl, destPath1, false, client.WithToken(tok))
 	require.NoError(t, err, "Should be able to download file through federation")
 	assert.Equal(t, int64(len(testFileContent)), transferResults[0].TransferredBytes)
 
 	content1, err := os.ReadFile(destPath1)
 	require.NoError(t, err)
 	assert.Equal(t, testFileContent, string(content1), "Downloaded content should match original")
 
-	// Get the cache server to manipulate its state
-	cacheServer := &cache.CacheServer{}
+	// Step 2: Simulate origin downtime by POSTing new origin ad without /test namespace
+	metadata, err := server_utils.GetServerMetadata(ctx, server_structs.OriginType)
+	require.NoError(t, err)
+
+	issuerUrlStr, err := config.GetServerIssuerURL()
+	require.NoError(t, err)
+	issuerUrl, err := url.Parse(issuerUrlStr)
+	require.NoError(t, err)
+
+	// Create advertisement with empty namespace list
+	emptyAd := server_structs.OriginAdvertiseV2{
+		ServerID:   metadata.ID,
+		DataURL:    param.Origin_Url.GetString(),
+		WebURL:     param.Server_ExternalWebUrl.GetString(),
+		Namespaces: []server_structs.NamespaceAdV2{},
+		Issuer: []server_structs.TokenIssuer{{
+			IssuerUrl: *issuerUrl,
+		}},
+		StorageType: server_structs.OriginStoragePosix,
+	}
+	emptyAd.Initialize(metadata.Name)
+	emptyAd.Now = time.Now()
+
+	body, err := json.Marshal(emptyAd)
+	require.NoError(t, err)
+
+	directorUrlStr := param.Server_ExternalWebUrl.GetString() + "/api/v1.0/director/registerOrigin"
+	directorUrl, err := url.Parse(directorUrlStr)
+	require.NoError(t, err)
+
+	// Create advertisement token
+	advTokenCfg := token.NewWLCGToken()
+	advTokenCfg.Lifetime = time.Minute
+	advTokenCfg.Issuer = issuerUrlStr
+	advTokenCfg.AddAudienceAny()
+	advTokenCfg.Subject = param.Server_Hostname.GetString()
+	advTokenCfg.AddScopes(token_scopes.Pelican_Advertise)
+	advTok, err := advTokenCfg.CreateToken()
+	require.NoError(t, err)
+
+	// POST the empty advertisement
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, directorUrl.String(), bytes.NewBuffer(body))
+	require.NoError(t, err)
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", "Bearer "+advTok)
+	req.Header.Set("User-Agent", "pelican-test/"+config.GetVersion())
+
+	tr := config.GetTransport()
+	httpClient := &http.Client{Transport: tr}
+	resp, err := httpClient.Do(req)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	require.Equal(t, http.StatusOK, resp.StatusCode, "Failed to register empty origin advertisement")
+
+	// Wait for director to process the empty advertisement and remove /test namespace
+	namespacesUrl := param.Server_ExternalWebUrl.GetString() + "/api/v1.0/director/listNamespaces"
+	require.Eventually(t, func() bool {
+		resp, err := httpClient.Get(namespacesUrl)
+		if err != nil {
+			return false
+		}
+		defer resp.Body.Close()
+
+		var namespaces []server_structs.NamespaceAdV2
+		if err := json.NewDecoder(resp.Body).Decode(&namespaces); err != nil {
+			return false
+		}
 
-	// Step 2: Simulate origin downtime by removing namespace ads from cache
-	cacheServer.SetNamespaceAds([]server_structs.NamespaceAdV2{})
+		// Check that /test namespace is no longer present
+		for _, ns := range namespaces {
+			if ns.Path == "/test" {
+				return false
+			}
+		}
+		return true
+	}, 5*time.Second, 100*time.Millisecond, "Director should remove /test namespace after processing empty advertisement")
 
 	// Step 3: Trigger cache authz refresh by overwriting Xrootd.ScitokensConfig with an unrelated issuer
+	// The file watcher will detect this change and call EmitScitokensConfig, which uses cached namespace ads
+	// from the cache's last GetNamespaceAdsFromDirector() call (which gets data from the director).
+	// Since we just updated the director to remove /test, the cache's cached ads should now reflect that.
 	unrelatedConfig := `
 [Global]
 audience = https://wlcg.cern.ch/jwt/v1/any
@@ -136,15 +232,20 @@ default_user = xrootd
 	err = os.WriteFile(scitokensConfigPath, []byte(unrelatedConfig), 0644)
 	require.NoError(t, err)
 
-	// Trigger the refresh by calling EmitScitokensConfig
-	err = xrootd.EmitScitokensConfig(cacheServer)
-	require.NoError(t, err)
-
-	// Verify the refresh happened by checking the generated file
+	// Wait for the background cache process to emit the config with /unrelated/path
 	generatedConfigPath := filepath.Join(param.Cache_RunLocation.GetString(), "scitokens-cache-generated.cfg")
-	generatedContent, err := os.ReadFile(generatedConfigPath)
-	require.NoError(t, err)
-	assert.Contains(t, string(generatedContent), "unrelated-issuer.example.com", "Generated config should contain the unrelated issuer")
+	require.Eventually(t, func() bool {
+		generatedContent, err := os.ReadFile(generatedConfigPath)
+		if err != nil {
+			return false
+		}
+		contentStr := string(generatedContent)
+		// Check that the unrelated issuer override is present and /test is gone
+		return len(contentStr) > 0 &&
+			strings.Contains(contentStr, "unrelated-issuer.example.com") &&
+			strings.Contains(contentStr, "/unrelated/path") &&
+			!strings.Contains(contentStr, "/test")
+	}, 10*time.Second, 100*time.Millisecond, "Generated config should contain unrelated issuer override and not /test")
 
 	// Step 4: Verify data is no longer accessible through the cache
 	// Try to download directly from cache - should fail because authorization is missing
@@ -166,21 +267,28 @@ default_user = xrootd
 	err = os.WriteFile(scitokensConfigPath, []byte(properConfig), 0644)
 	require.NoError(t, err)
 
-	// Trigger the refresh again
-	err = xrootd.EmitScitokensConfig(cacheServer)
-	require.NoError(t, err)
-
-	// Verify the refresh happened by checking the generated file
-	generatedContent, err = os.ReadFile(generatedConfigPath)
-	require.NoError(t, err)
-	assert.Contains(t, string(generatedContent), serverIssuerUrl, "Generated config should contain the server issuer")
-	assert.Contains(t, string(generatedContent), "/test", "Generated config should contain the /test base path")
+	// Wait for the background cache process to emit the updated config with /test
+	require.Eventually(t, func() bool {
+		generatedContent, err := os.ReadFile(generatedConfigPath)
+		if err != nil {
+			return false
+		}
+		contentStr := string(generatedContent)
+		// Check that the unrelated issuer is gone and our server issuer is present
+		return len(contentStr) > 0 &&
+			!strings.Contains(contentStr, "unrelated-issuer.example.com") &&
+			strings.Contains(contentStr, serverIssuerUrl) &&
+			strings.Contains(contentStr, "TestIssuer")
+	}, 10*time.Second, 100*time.Millisecond, "Generated config should contain server issuer and not unrelated issuer")
 
-	// Step 6: Verify cached object is now accessible even with origin "offline"
+	// Step 6: Verify cached object is accessible with proper auth, even with origin "offline"
+	// Use client.DoGet with WithCaches to force use of specific cache
 	destPath3 := filepath.Join(tmpDir, "downloaded3.txt")
-	transferResults3, err := client.DoGet(ctx, cacheUrl+"/test/"+testFileName, destPath3, false, client.WithToken(tok))
+	cacheUrlParsed, err := url.Parse(cacheUrl)
+	require.NoError(t, err)
+	transferResults, err = client.DoGet(ctx, pelicanUrl, destPath3, false, client.WithToken(tok), client.WithCaches(cacheUrlParsed))
 	require.NoError(t, err, "Should be able to access cached data with proper authorization")
-	assert.Equal(t, int64(len(testFileContent)), transferResults3[0].TransferredBytes)
+	assert.Equal(t, int64(len(testFileContent)), transferResults[0].TransferredBytes)
 
 	content3, err := os.ReadFile(destPath3)
 	require.NoError(t, err)