Add a local cache module

broker/broker_test.go

@@ -32,6 +32,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pelicanplatform/pelican/common"
 	"github.com/pelicanplatform/pelican/config"
 	"github.com/pelicanplatform/pelican/param"
 	"github.com/pelicanplatform/pelican/registry"
@@ -307,7 +308,7 @@ func TestRetrieveTimeout(t *testing.T) {
 	err = json.Unmarshal(responseBytes, &brokerResp)
 	require.NoError(t, err)
 
-	assert.Equal(t, brokerReponseStatusTimeout, brokerResp.Status)
+	assert.Equal(t, common.RespPollTimeout, brokerResp.Status)
 
 	ctx, cancelFunc := context.WithTimeout(ctx, 50*time.Millisecond)
 	defer cancelFunc()

broker/client.go

@@ -43,6 +43,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/pelicanplatform/pelican/common"
 	"github.com/pelicanplatform/pelican/config"
 	"github.com/pelicanplatform/pelican/param"
 	"github.com/pelicanplatform/pelican/token_scopes"
@@ -239,7 +240,7 @@ func ConnectToOrigin(ctx context.Context, brokerUrl, prefix, originName string)
 		err = errors.Wrap(err, "Failure when reading response from broker response")
 	}
 	if resp.StatusCode >= 400 {
-		errResp := brokerMsgResp{}
+		errResp := common.SimpleApiResp{}
 		log.Errorf("Failure (status code %d) when invoking the broker: %s", resp.StatusCode, string(responseBytes))
 		if err = json.Unmarshal(responseBytes, &errResp); err != nil {
 			err = errors.Errorf("Failure when invoking the broker (status code %d); unable to parse error message", resp.StatusCode)
@@ -298,7 +299,7 @@ func ConnectToOrigin(ctx context.Context, brokerUrl, prefix, originName string)
 		hj, ok := writer.(http.Hijacker)
 		if !ok {
 			log.Debug("Not able to hijack underlying TCP connection from server")
-			resp := brokerMsgResp{
+			resp := common.SimpleApiResp{
 				Msg:    "Unable to reverse TCP connection; HTTP/2 in use",
 				Status: "error",
 			}
@@ -469,7 +470,7 @@ func doCallback(ctx context.Context, brokerResp reversalRequest) (listener net.L
 	}
 
 	if resp.StatusCode >= 400 {
-		errResp := brokerMsgResp{}
+		errResp := common.SimpleApiResp{}
 		if err = json.Unmarshal(responseBytes, &errResp); err != nil {
 			err = errors.Errorf("Failure when invoking cache %s callback (status code %d); unable to parse error message", brokerResp.CallbackUrl, resp.StatusCode)
 		} else {
@@ -616,7 +617,7 @@ func LaunchRequestMonitor(ctx context.Context, egrp *errgroup.Group, resultChan
 					break
 				}
 				if resp.StatusCode >= 400 {
-					errResp := brokerMsgResp{}
+					errResp := common.SimpleApiResp{}
 					if err = json.Unmarshal(responseBytes, &errResp); err != nil {
 						log.Errorf("Failure when invoking the broker (status code %d); unable to parse error message", resp.StatusCode)
 					} else {
@@ -633,17 +634,17 @@ func LaunchRequestMonitor(ctx context.Context, egrp *errgroup.Group, resultChan
 					break
 				}
 
-				if brokerResp.Status == brokerResponseStatusOK {
+				if brokerResp.Status == common.RespOK {
 					listener, err := doCallback(ctx, brokerResp.Request)
 					if err != nil {
 						log.Errorln("Failed to callback to the cache:", err)
 						resultChan <- err
 						break
 					}
 					resultChan <- listener
-				} else if brokerResp.Status == brokerReponseStatusFailed {
+				} else if brokerResp.Status == common.RespFailed {
 					log.Errorln("Broker responded to origin retrieve with an error:", brokerResp.Msg)
-				} else if brokerResp.Status != brokerReponseStatusTimeout { // We expect timeouts; do not log them.
+				} else if brokerResp.Status != common.RespPollTimeout { // We expect timeouts; do not log them.
 					if brokerResp.Msg != "" {
 						log.Errorf("Broker responded with unknown status (%s); msg: %s", brokerResp.Status, brokerResp.Msg)
 					} else {

broker/server_apis.go

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/gin-gonic/gin"
+	"github.com/pelicanplatform/pelican/common"
 	"github.com/pelicanplatform/pelican/param"
 	"github.com/pelicanplatform/pelican/token_scopes"
 	"github.com/pkg/errors"
@@ -39,17 +40,9 @@ type (
 		Prefix string `json:"prefix"`
 	}
 
-	brokerResponseStatus string
-
-	// Base response for a request or retrieval
-	brokerMsgResp struct {
-		Status brokerResponseStatus `json:"status"`
-		Msg    string               `json:"msg,omitempty"`
-	}
-
 	// Response for a successful retrieval
 	brokerRetrievalResp struct {
-		brokerMsgResp
+		common.SimpleApiResp
 		Request reversalRequest `json:"req"`
 	}
 
@@ -59,27 +52,21 @@ type (
 	}
 )
 
-const (
-	brokerResponseStatusOK     brokerResponseStatus = "success"
-	brokerReponseStatusFailed  brokerResponseStatus = "error"
-	brokerReponseStatusTimeout brokerResponseStatus = "timeout"
-)
-
 func newBrokerReqResp(req reversalRequest) (result brokerRetrievalResp) {
 	result.Request = req
-	result.brokerMsgResp.Status = brokerResponseStatusOK
+	result.SimpleApiResp.Status = common.RespOK
 	return
 }
 
-func newBrokerRespFail(msg string) brokerMsgResp {
-	return brokerMsgResp{
-		Status: brokerReponseStatusFailed,
+func newBrokerRespFail(msg string) common.SimpleApiResp {
+	return common.SimpleApiResp{
+		Status: common.RespFailed,
 		Msg:    msg,
 	}
 }
 
 func newBrokerRespTimeout() (result brokerRetrievalResp) {
-	result.brokerMsgResp.Status = brokerReponseStatusTimeout
+	result.SimpleApiResp.Status = common.RespPollTimeout
 	return
 }
 

broker/token_utils.go

@@ -123,15 +123,12 @@ func createToken(namespace, subject, audience string, desiredScope token_scopes.
 		return
 	}
 
-	tokenCfg := token.TokenConfig{
-		Lifetime:     time.Minute,
-		TokenProfile: token.WLCG,
-		Audience:     []string{audience},
-		Issuer:       issuerUrl,
-		Version:      "1.0",
-		Subject:      subject,
-	}
-	tokenCfg.AddScopes([]token_scopes.TokenScope{desiredScope})
+	tokenCfg := token.NewWLCGToken()
+	tokenCfg.Lifetime = time.Minute
+	tokenCfg.Issuer = issuerUrl
+	tokenCfg.Subject = subject
+	tokenCfg.AddAudiences(audience)
+	tokenCfg.AddScopes(desiredScope)
 	tokenStr, err = tokenCfg.CreateToken()
 
 	return
@@ -170,7 +167,7 @@ func verifyToken(ctx context.Context, token, namespace, audience string, require
 
 	scopeValidator := token_scopes.CreateScopeValidator([]token_scopes.TokenScope{requiredScope}, false)
 	err = jwt.Validate(tok,
-		jwt.WithAudience(param.Server_ExternalWebUrl.GetString()),
+		jwt.WithAudience(audience),
 		jwt.WithValidator(scopeValidator),
 		jwt.WithClaimValue("iss", issuerUrl),
 	)

cache_ui/advertise.go

@@ -19,6 +19,7 @@
 package cache_ui
 
 import (
+	"context"
 	"encoding/json"
 	"net/url"
 	"strings"
@@ -63,10 +64,6 @@ func (server *CacheServer) GetNamespaceAdsFromDirector() error {
 		return errors.Wrap(err, "Unable to parse director url")
 	}
 
-	if err != nil {
-		return errors.Wrapf(err, "Failed to get DirectorURL from config: %v", err)
-	}
-
 	// Create the listNamespaces url
 	directorNSListEndpointURL, err := url.JoinPath(directorEndpointURL.String(), "api", "v2.0", "director", "listNamespaces")
 	if err != nil {
@@ -76,14 +73,14 @@ func (server *CacheServer) GetNamespaceAdsFromDirector() error {
 	// Attempt to get data from the 2.0 endpoint, if that returns a 404 error, then attempt to get data
 	// from the 1.0 endpoint and convert from V1 to V2
 
-	respData, err := utils.MakeRequest(directorNSListEndpointURL, "GET", nil, nil)
+	respData, err := utils.MakeRequest(context.Background(), directorNSListEndpointURL, "GET", nil, nil)
 	if err != nil {
 		if strings.Contains(err.Error(), "404") {
 			directorNSListEndpointURL, err = url.JoinPath(directorEndpoint, "api", "v1.0", "director", "listNamespaces")
 			if err != nil {
 				return err
 			}
-			respData, err = utils.MakeRequest(directorNSListEndpointURL, "GET", nil, nil)
+			respData, err = utils.MakeRequest(context.Background(), directorNSListEndpointURL, "GET", nil, nil)
 			var respNSV1 []common.NamespaceAdV1
 			if err != nil {
 				return errors.Wrap(err, "Failed to make request")

client/director.go

@@ -23,6 +23,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"path"
 	"sort"
 	"strconv"
 	"strings"
@@ -234,7 +235,7 @@ func GetCachesFromDirectorResponse(resp *http.Response, needsToken bool) (caches
 }
 
 // NewTransferDetails creates the TransferDetails struct with the given cache
-func NewTransferDetailsUsingDirector(cache namespaces.DirectorCache, opts transferDetailsOptions) []transferAttemptDetails {
+func newTransferDetailsUsingDirector(cache namespaces.DirectorCache, opts transferDetailsOptions) []transferAttemptDetails {
 	details := make([]transferAttemptDetails, 0)
 	cacheEndpoint := cache.EndpointUrl
 
@@ -244,7 +245,9 @@ func NewTransferDetailsUsingDirector(cache namespaces.DirectorCache, opts transf
 		log.Errorln("Failed to parse cache:", cache, "error:", err)
 		return nil
 	}
-	if cacheURL.Host == "" {
+	if cacheURL.Scheme == "unix" && cacheURL.Host != "" {
+		cacheURL.Path = path.Clean("/" + path.Join(cacheURL.Host, cacheURL.Path))
+	} else if cacheURL.Scheme != "unix" && cacheURL.Host == "" {
 		// Assume the cache is just a hostname
 		cacheURL.Host = cacheEndpoint
 		cacheURL.Path = ""
@@ -253,26 +256,35 @@ func NewTransferDetailsUsingDirector(cache namespaces.DirectorCache, opts transf
 	}
 	log.Debugf("Parsed Cache: %s", cacheURL.String())
 	if opts.NeedsToken {
-		cacheURL.Scheme = "https"
-		if !hasPort(cacheURL.Host) {
-			// Add port 8444 and 8443
-			urlCopy := *cacheURL
-			urlCopy.Host += ":8444"
-			details = append(details, transferAttemptDetails{
-				Url:        &urlCopy,
-				Proxy:      false,
-				PackOption: opts.PackOption,
-			})
-			// Strip the port off and add 8443
-			cacheURL.Host = cacheURL.Host + ":8443"
+		// Unless we're using the local Unix domain socket cache, force HTTPS
+		if cacheURL.Scheme != "unix" {
+			cacheURL.Scheme = "https"
+			if !hasPort(cacheURL.Host) {
+				// Add port 8444 and 8443
+				urlCopy := *cacheURL
+				urlCopy.Host += ":8444"
+				details = append(details, transferAttemptDetails{
+					Url:        &urlCopy,
+					Proxy:      false,
+					PackOption: opts.PackOption,
+				})
+				// Strip the port off and add 8443
+				cacheURL.Host = cacheURL.Host + ":8443"
+			}
 		}
-		// Whether port is specified or not, add a transfer without proxy
-		details = append(details, transferAttemptDetails{
+		det := transferAttemptDetails{
 			Url:        cacheURL,
 			Proxy:      false,
 			PackOption: opts.PackOption,
-		})
-	} else {
+		}
+		if cacheURL.Scheme == "unix" {
+			det.UnixSocket = cacheURL.Path
+		}
+		// Whether port is specified or not, add a transfer without proxy
+		details = append(details, det)
+	} else if cacheURL.Scheme == "" || cacheURL.Scheme == "http" {
+		// Assume a transfer not needing a token and not specifying a scheme is HTTP
+		// WARNING: This is legacy code; we should always specify a scheme
 		cacheURL.Scheme = "http"
 		if !hasPort(cacheURL.Host) {
 			cacheURL.Host += ":8000"
@@ -290,6 +302,18 @@ func NewTransferDetailsUsingDirector(cache namespaces.DirectorCache, opts transf
 				PackOption: opts.PackOption,
 			})
 		}
+	} else {
+		// A non-HTTP scheme is specified and a token is not needed; this wasn't possible
+		// in the legacy cases.  Simply use the provided config
+		det := transferAttemptDetails{
+			Url:        cacheURL,
+			Proxy:      false,
+			PackOption: opts.PackOption,
+		}
+		if cacheURL.Scheme == "unix" {
+			det.UnixSocket = cacheURL.Path
+		}
+		details = append(details, det)
 	}
 
 	return details

client/director_test.go

@@ -155,7 +155,7 @@ func TestNewTransferDetailsUsingDirector(t *testing.T) {
 
 	// Case 1: cache with http
 
-	transfers := NewTransferDetailsUsingDirector(nonAuthCache, transferDetailsOptions{nonAuthCache.AuthedReq, ""})
+	transfers := newTransferDetailsUsingDirector(nonAuthCache, transferDetailsOptions{nonAuthCache.AuthedReq, ""})
 	assert.Equal(t, 2, len(transfers))
 	assert.Equal(t, "my-cache-url:8000", transfers[0].Url.Host)
 	assert.Equal(t, "http", transfers[0].Url.Scheme)
@@ -166,15 +166,15 @@ func TestNewTransferDetailsUsingDirector(t *testing.T) {
 	assert.Equal(t, false, transfers[1].Proxy)
 
 	// Case 2: cache with https
-	transfers = NewTransferDetailsUsingDirector(authCache, transferDetailsOptions{authCache.AuthedReq, ""})
+	transfers = newTransferDetailsUsingDirector(authCache, transferDetailsOptions{authCache.AuthedReq, ""})
 	assert.Equal(t, 1, len(transfers))
 	assert.Equal(t, "my-cache-url:8443", transfers[0].Url.Host)
 	assert.Equal(t, "https", transfers[0].Url.Scheme)
 	assert.Equal(t, false, transfers[0].Proxy)
 
 	// Case 3: cache without port with http
 	nonAuthCache.EndpointUrl = "my-cache-url"
-	transfers = NewTransferDetailsUsingDirector(nonAuthCache, transferDetailsOptions{nonAuthCache.AuthedReq, ""})
+	transfers = newTransferDetailsUsingDirector(nonAuthCache, transferDetailsOptions{nonAuthCache.AuthedReq, ""})
 	assert.Equal(t, 2, len(transfers))
 	assert.Equal(t, "my-cache-url:8000", transfers[0].Url.Host)
 	assert.Equal(t, "http", transfers[0].Url.Scheme)
@@ -185,7 +185,7 @@ func TestNewTransferDetailsUsingDirector(t *testing.T) {
 
 	// Case 4. cache without port with https
 	authCache.EndpointUrl = "my-cache-url"
-	transfers = NewTransferDetailsUsingDirector(authCache, transferDetailsOptions{authCache.AuthedReq, ""})
+	transfers = newTransferDetailsUsingDirector(authCache, transferDetailsOptions{authCache.AuthedReq, ""})
 	assert.Equal(t, 2, len(transfers))
 	assert.Equal(t, "my-cache-url:8444", transfers[0].Url.Host)
 	assert.Equal(t, "https", transfers[0].Url.Scheme)