Send correct logs to Saturn logger

caboose.go

@@ -45,6 +45,9 @@ type Config struct {
 	PoolLowWatermark int
 	// MaxRetrievalAttempts determines the number of times we will attempt to retrieve a block from the Saturn network before failing.
 	MaxRetrievalAttempts int
+
+	// SaturnLoggerJWT is the JWT Auth token to use when submitting logs to the Saturn logging endpoint.
+	SaturnLoggerJWT string
 }
 
 const DefaultMaxRetries = 3
@@ -65,6 +68,10 @@ type Caboose struct {
 }
 
 func NewCaboose(config *Config) (ipfsblockstore.Blockstore, error) {
+	if config.SaturnLoggerJWT == "" || len(config.SaturnLoggerJWT) == 0 {
+		return nil, errors.New("JWT token required for Saturn Logger")
+	}
+
 	c := Caboose{
 		config: config,
 		pool:   newPool(config),

cmd/caboose/main.go

@@ -56,9 +56,10 @@ func main1() int {
 				LoggingClient:   http.DefaultClient,
 				LoggingInterval: 5 * time.Second,
 
-				DoValidation: true,
-				PoolRefresh:  5 * time.Minute,
-				SaturnClient: &saturnClient,
+				DoValidation:    true,
+				PoolRefresh:     5 * time.Minute,
+				SaturnClient:    &saturnClient,
+				SaturnLoggerJWT: "test",
 			})
 			if err != nil {
 				return err
@@ -84,5 +85,6 @@ func main1() int {
 		log.Println(err)
 		return 1
 	}
+	time.Sleep(5 * time.Second)
 	return 0
 }

failure_test.go

@@ -59,6 +59,7 @@ func TestCabooseFailures(t *testing.T) {
 		PoolWeightChangeDebounce: time.Duration(1),
 		PoolRefresh:              time.Millisecond * 50,
 		MaxRetrievalAttempts:     2,
+		SaturnLoggerJWT:          "rand",
 	})
 	if err != nil {
 		t.Fatal(err)

log.go

@@ -3,6 +3,7 @@ package caboose
 import (
 	"bytes"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"net/url"
 	"time"
@@ -19,6 +20,7 @@ type logger struct {
 	client   *http.Client
 	endpoint url.URL
 	done     chan struct{}
+	jwt      string
 }
 
 func newLogger(c *Config) *logger {
@@ -28,6 +30,7 @@ func newLogger(c *Config) *logger {
 		client:   c.LoggingClient,
 		endpoint: c.LoggingEndpoint,
 		done:     make(chan struct{}),
+		jwt:      c.SaturnLoggerJWT,
 	}
 	go l.background()
 	return &l
@@ -59,7 +62,24 @@ func (l *logger) submit(logs []log) {
 	finalLogs := bytes.NewBuffer(nil)
 	enc := json.NewEncoder(finalLogs)
 	enc.Encode(logBatch{logs})
-	l.client.Post(l.endpoint.String(), "application/json", finalLogs)
+
+	req, err := http.NewRequest(http.MethodPost, l.endpoint.String(), finalLogs)
+	if err != nil {
+		goLogger.Errorw("failed to create http request to submit saturn logs", "err", err)
+		return
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", l.jwt))
+
+	resp, err := l.client.Do(req)
+	if err != nil {
+		goLogger.Errorw("failed to submit saturn logs", "err", err)
+		return
+	}
+	if resp.StatusCode != http.StatusOK {
+		goLogger.Errorw("saturn logging endpoint did not return 200", "status", resp.StatusCode)
+	}
 }
 
 func (l *logger) Close() {
@@ -71,17 +91,19 @@ type logBatch struct {
 }
 
 type log struct {
-	CacheHit        bool      `json:"cacheHit"`
-	URL             string    `json:"url"`
-	LocalTime       time.Time `json:"localTime"`
-	NumBytesSent    int       `json:"numBytesSent"`
-	RequestDuration float64   `json:"requestDuration"` // in seconds
-	RequestID       string    `json:"requestId"`
-	HTTPStatusCode  int       `json:"httpStatusCode"`
-	HTTPProtocol    string    `json:"httpProtocol"`
-	TTFBMS          int       `json:"ttfbMs"`
-	ClientAddress   string    `json:"clientAddress"`
-	Range           string    `json:"range"`
-	Referrer        string    `json:"referrer"`
-	UserAgent       string    `json:"userAgent"`
+	CacheHit           bool      `json:"cacheHit"`
+	URL                string    `json:"url"`
+	StartTime          time.Time `json:"startTime"`
+	NumBytesSent       int       `json:"numBytesSent"`
+	RequestDurationSec float64   `json:"requestDurationSec"` // in seconds
+	RequestID          string    `json:"requestId"`
+	HTTPStatusCode     int       `json:"httpStatusCode"`
+	HTTPProtocol       string    `json:"httpProtocol"`
+	TTFBMS             int       `json:"ttfbMs"`
+	Range              string    `json:"range"`
+	Referrer           string    `json:"referrer"`
+	UserAgent          string    `json:"userAgent"`
+	NodeId             string    `json:"nodeId"`
+	IfNetworkError     string    `json:"ifNetworkError"`
+	NodeIpAddress      string    `json:"nodeIpAddress"`
 }

pool.go

@@ -341,6 +341,13 @@ func (p *pool) updateWeightUnlocked(node string, failure bool) (index int, membe
 
 var saturnReqTmpl = "https://%s/ipfs/%s?format=raw"
 
+var (
+	saturnNodeIdKey     = "Saturn-Node-Id"
+	saturnTransferIdKey = "Saturn-Transfer-Id"
+	saturnCacheHitKey   = "Saturn-Cache-Status"
+	saturnCacheHit      = "HIT"
+)
+
 // doFetch attempts to fetch a block from a given Saturn endpoint. It sends the retrieval logs to the logging endpoint upon a successful or failed attempt.
 func (p *pool) doFetch(ctx context.Context, from string, c cid.Cid, attempt int) (b blocks.Block, e error) {
 	requestId := uuid.NewString()
@@ -351,6 +358,13 @@ func (p *pool) doFetch(ctx context.Context, from string, c cid.Cid, attempt int)
 	proto := "unknown"
 	respReq := &http.Request{}
 	received := 0
+	reqUrl := ""
+	var respHeader http.Header
+	saturnNodeId := ""
+	saturnTransferId := ""
+	isCacheHit := false
+	networkError := ""
+
 	defer func() {
 		ttfbMs := fb.Sub(start).Milliseconds()
 		durationSecs := time.Since(start).Seconds()
@@ -363,28 +377,45 @@ func (p *pool) doFetch(ctx context.Context, from string, c cid.Cid, attempt int)
 			fetchSpeedMetric.Observe(float64(received) / durationSecs)
 			fetchSizeMetric.Observe(float64(received))
 		}
+
+		if respHeader != nil {
+			saturnNodeId = respHeader.Get(saturnNodeIdKey)
+			saturnTransferId = respHeader.Get(saturnTransferIdKey)
+
+			cacheHit := respHeader.Get(saturnCacheHitKey)
+			if cacheHit == saturnCacheHit {
+				isCacheHit = true
+			}
+
+			for k, v := range respHeader {
+				received = received + len(k) + len(v)
+			}
+		}
+
 		p.logger.queue <- log{
-			CacheHit:  false,
-			URL:       from,
-			LocalTime: start,
-			// TODO: does this include header sizes?
-			NumBytesSent:    received,
-			RequestDuration: durationSecs,
-			RequestID:       requestId,
-			HTTPStatusCode:  code,
-			HTTPProtocol:    proto,
-			TTFBMS:          int(ttfbMs),
+			CacheHit:           isCacheHit,
+			URL:                reqUrl,
+			StartTime:          start,
+			NumBytesSent:       received,
+			RequestDurationSec: durationSecs,
+			RequestID:          saturnTransferId,
+			HTTPStatusCode:     code,
+			HTTPProtocol:       proto,
+			TTFBMS:             int(ttfbMs),
 			// my address
-			ClientAddress: "",
-			Range:         "",
-			Referrer:      respReq.Referer(),
-			UserAgent:     respReq.UserAgent(),
+			Range:          "",
+			Referrer:       respReq.Referer(),
+			UserAgent:      respReq.UserAgent(),
+			NodeId:         saturnNodeId,
+			NodeIpAddress:  from,
+			IfNetworkError: networkError,
 		}
 	}()
 
 	reqCtx, cancel := context.WithTimeout(ctx, DefaultSaturnRequestTimeout)
 	defer cancel()
-	req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, fmt.Sprintf(saturnReqTmpl, from, c), nil)
+	reqUrl = fmt.Sprintf(saturnReqTmpl, from, c)
+	req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, reqUrl, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -401,8 +432,10 @@ func (p *pool) doFetch(ctx context.Context, from string, c cid.Cid, attempt int)
 	resp, err := p.config.SaturnClient.Do(req)
 	fb = time.Now()
 	if err != nil {
+		networkError = err.Error()
 		return nil, fmt.Errorf("http request failed: %w", err)
 	}
+	respHeader = resp.Header
 	defer resp.Body.Close()
 
 	code = resp.StatusCode