From ae08d035c89198630580719addb229114c4fa538 Mon Sep 17 00:00:00 2001 From: Kevin Franklin Kim Date: Thu, 30 Jul 2026 17:15:46 +0200 Subject: [PATCH] feat: enable golangci-lint wsl_5 & whitespace --- .golangci.yaml | 4 +-- .goreleaser.yaml | 3 +- client/client.go | 9 +++++ client/client_test.go | 19 +++++++++++ client/connectionpool.go | 13 ++++++++ client/httptransport.go | 5 +++ client/httptransport_test.go | 3 ++ client/sockettransport.go | 13 ++++++++ client/sockettransport_test.go | 2 ++ cmd/http.go | 8 +++++ cmd/root.go | 1 + cmd/socket.go | 5 +++ cmd/version.go | 1 + content/reponode.go | 8 +++++ pkg/handler/http.go | 8 +++++ pkg/handler/socket.go | 23 +++++++++++++ pkg/metrics/metrics.go | 3 ++ pkg/repo/history.go | 9 +++++ pkg/repo/history_test.go | 10 ++++++ pkg/repo/loader.go | 47 ++++++++++++++++++++++++++ pkg/repo/loader_test.go | 27 +++++++++++---- pkg/repo/mock/mock.go | 2 ++ pkg/repo/repo.go | 60 +++++++++++++++++++++++++++++++++- pkg/repo/repo_test.go | 11 +++++++ pkg/repo/storage_blob.go | 16 +++++++++ pkg/repo/storage_blob_test.go | 4 +++ pkg/repo/storage_fs.go | 9 +++++ pkg/repo/storage_fs_test.go | 2 ++ 28 files changed, 312 insertions(+), 13 deletions(-) diff --git a/.golangci.yaml b/.golangci.yaml index 4afee49..b4c7582 100644 --- a/.golangci.yaml +++ b/.golangci.yaml @@ -4,7 +4,7 @@ version: "2" run: go: "1.26" - build-tags: [ safe ] + build-tags: [ safe, pprof ] modules-download-mode: readonly linters: @@ -16,8 +16,6 @@ linters: - paralleltest - godoclint - forbidigo - - wsl_v5 - - whitespace # Discouraged linters - noinlineerr # Disallows inline error handling (`if err := ...; err != nil {`). - embeddedstructfieldcheck # Embedded types should be at the top of the field list of a struct, and there must be an empty line separating embedded fields from regular fields. [fast] diff --git a/.goreleaser.yaml b/.goreleaser.yaml index 93b5c45..38ffb50 100644 --- a/.goreleaser.yaml +++ b/.goreleaser.yaml @@ -25,8 +25,7 @@ builds: - '7' flags: - -trimpath - - -tags=safe - - -tags=pprof + - -tags=safe,pprof ldflags: - -s -w -X github.com/foomo/contentserver/cmd.version={{.Version}} diff --git a/client/client.go b/client/client.go index 9f428eb..4968b76 100644 --- a/client/client.go +++ b/client/client.go @@ -39,10 +39,12 @@ func (c *Client) Update(ctx context.Context) (*responses.Update, error) { type serverResponse struct { Reply *responses.Update } + resp := serverResponse{} if err := c.t.Call(ctx, handler.RouteUpdate, &requests.Update{}, &resp); err != nil { return nil, err } + return resp.Reply, nil } @@ -51,6 +53,7 @@ func (c *Client) GetContent(ctx context.Context, request *requests.Content) (*co type serverResponse struct { Reply *content.SiteContent } + resp := serverResponse{} if err := c.t.Call(ctx, handler.RouteGetContent, request, &resp); err != nil { return nil, err @@ -69,6 +72,7 @@ func (c *Client) GetURIs(ctx context.Context, dimension string, ids []string) (m if err := c.t.Call(ctx, handler.RouteGetURIs, &requests.URIs{Dimension: dimension, IDs: ids}, &resp); err != nil { return nil, err } + return resp.Reply, nil } @@ -78,13 +82,16 @@ func (c *Client) GetNodes(ctx context.Context, env *requests.Env, nodes map[stri Env: env, Nodes: nodes, } + type serverResponse struct { Reply map[string]*content.Node } + resp := serverResponse{} if err := c.t.Call(ctx, handler.RouteGetNodes, r, &resp); err != nil { return nil, err } + return resp.Reply, nil } @@ -93,10 +100,12 @@ func (c *Client) GetRepo(ctx context.Context) (map[string]*content.RepoNode, err type serverResponse struct { Reply map[string]*content.RepoNode } + resp := serverResponse{} if err := c.t.Call(ctx, handler.RouteGetRepo, &requests.Repo{}, &resp); err != nil { return nil, err } + return resp.Reply, nil } diff --git a/client/client_test.go b/client/client_test.go index 3132151..664203a 100644 --- a/client/client_test.go +++ b/client/client_test.go @@ -32,8 +32,10 @@ func TestGetURIs(t *testing.T) { testWithClients(t, func(t *testing.T, c *client.Client) { t.Helper() t.Parallel() + request := mock.MakeValidURIsRequest() uriMap, err := c.GetURIs(t.Context(), request.Dimension, request.IDs) + time.Sleep(100 * time.Millisecond) require.NoError(t, err) assert.Equal(t, "/a", uriMap[request.IDs[0]]) @@ -46,6 +48,7 @@ func TestGetRepo(t *testing.T) { t.Parallel() r, err := c.GetRepo(t.Context()) require.NoError(t, err) + if assert.NotEmpty(t, r, "received empty JSON from GetRepo") { assert.InDelta(t, 1.0, r["dimension_foo"].Nodes["id-a"].Data["baz"].(float64), 0, "failed to drill deep for data") //nolint:forcetypeassert } @@ -56,17 +59,21 @@ func TestGetNodes(t *testing.T) { testWithClients(t, func(t *testing.T, c *client.Client) { t.Helper() t.Parallel() + nodesRequest := mock.MakeNodesRequest() nodes, err := c.GetNodes(t.Context(), nodesRequest.Env, nodesRequest.Nodes) require.NoError(t, err) + testNode, ok := nodes["test"] if !ok { t.Fatal("that should be a node") } + testData, ok := testNode.Item.Data["foo"] if !ok { t.Fatal("where is foo") } + if testData != "bar" { t.Fatal("testData should have bennd bar not", testData) } @@ -77,6 +84,7 @@ func TestGetContent(t *testing.T) { testWithClients(t, func(t *testing.T, c *client.Client) { t.Helper() t.Parallel() + request := mock.MakeValidContentRequest() response, err := c.GetContent(t.Context(), request) require.NoError(t, err) @@ -88,9 +96,12 @@ func TestGetContent(t *testing.T) { func benchmarkServerAndClientGetContent(b *testing.B, numGroups, numCalls int, client GetContentClient) { b.Helper() b.ResetTimer() + for i := 0; i < b.N; i++ { start := time.Now() + benchmarkClientAndServerGetContent(b, numGroups, numCalls, client) + dur := time.Since(start) totalCalls := numGroups * numCalls b.Log("requests per second", int(float64(totalCalls)/(float64(dur)/float64(1000000000))), dur, totalCalls) @@ -99,11 +110,14 @@ func benchmarkServerAndClientGetContent(b *testing.B, numGroups, numCalls int, c func benchmarkClientAndServerGetContent(tb testing.TB, numGroups, numCalls int, client GetContentClient) { tb.Helper() + var wg sync.WaitGroup wg.Add(numGroups) + for range numGroups { go func() { defer wg.Done() + request := mock.MakeValidContentRequest() for range numCalls { response, err := client.GetContent(tb.Context(), request) @@ -111,6 +125,7 @@ func benchmarkClientAndServerGetContent(tb testing.TB, numGroups, numCalls int, if request.URI != response.URI { tb.Fatal("uri mismatch") } + if response.Status != content.StatusOk { tb.Fatal("unexpected status") } @@ -149,17 +164,20 @@ func testWithClients(t *testing.T, testFunc func(t *testing.T, c *client.Client) func initRepo(tb testing.TB, l *zap.Logger) *repo.Repo { tb.Helper() testRepoServer, varDir := mock.GetMockData(tb) + h, err := repo.NewHistory(l, repo.HistoryWithHistoryDir(varDir), ) if err != nil { tb.Fatal(err) } + r := repo.New(l, testRepoServer.URL+"/repo-two-dimensions.json", h, ) up := make(chan bool, 1) + r.OnLoaded(func() { up <- true }) @@ -168,6 +186,7 @@ func initRepo(tb testing.TB, l *zap.Logger) *repo.Repo { // preventing race conditions with logging after test completion. ctx, cancel := context.WithCancel(context.Background()) go r.Start(ctx) //nolint:errcheck + <-up tb.Cleanup(func() { diff --git a/client/connectionpool.go b/client/connectionpool.go index 07ab93b..c56e97a 100644 --- a/client/connectionpool.go +++ b/client/connectionpool.go @@ -22,6 +22,7 @@ func newConnectionPool(url string, connectionPoolSize int, waitTimeout time.Dura chanDrainPool: make(chan int), } go connPool.run(connectionPoolSize, waitTimeout) + return connPool } @@ -31,6 +32,7 @@ func (c *connectionPool) run(connectionPoolSize int, waitTimeout time.Duration) err error conn net.Conn } + type waitPoolEntry struct { entryTime time.Time chanConn chan net.Conn @@ -46,6 +48,7 @@ func (c *connectionPool) run(connectionPoolSize int, waitTimeout time.Duration) busy: false, } } + RunLoop: for { // fmt.Println("----------------------- run loop ------------------------") @@ -55,6 +58,7 @@ RunLoop: for _, waitPoolEntry := range waitPool { waitPoolEntry.chanConn <- nil } + break RunLoop case <-time.After(waitTimeout): // fmt.Println("tick", len(connectionPool), len(waitPool)) @@ -72,6 +76,7 @@ RunLoop: nextI = i + 1 } } + waitPool[nextI] = &waitPoolEntry{ chanConn: chanReturnNextConn, entryTime: time.Now(), @@ -94,6 +99,7 @@ RunLoop: for _, poolEntry := range connectionPool { if poolEntry.conn == nil { var d net.Dialer + newConn, errDial := d.DialContext(context.Background(), "tcp", c.url) poolEntry.err = errDial poolEntry.conn = newConn @@ -104,12 +110,16 @@ RunLoop: if len(waitPool) == 0 { break } + if poolEntry.err == nil && poolEntry.conn != nil && !poolEntry.busy { for i, waitPoolEntry := range waitPool { // fmt.Println("---------------------------> serving wait pool", i, waitPoolEntry) poolEntry.busy = true + delete(waitPool, i) + waitPoolEntry.chanConn <- poolEntry.conn + break } } @@ -122,13 +132,16 @@ RunLoop: for i, waitPoolEntry := range waitPool { if now.Sub(waitPoolEntry.entryTime) > waitTimeout { waitPoolLoosers = append(waitPoolLoosers, i) + waitPoolEntry.chanConn <- nil } } + for _, i := range waitPoolLoosers { delete(waitPool, i) } } + c.chanDrainPool = nil c.chanConnReturn = nil c.chanConnGet = nil diff --git a/client/httptransport.go b/client/httptransport.go index 210231a..4352bcc 100644 --- a/client/httptransport.go +++ b/client/httptransport.go @@ -72,6 +72,7 @@ func (t *HTTPTransport) Call(ctx context.Context, route handler.Route, request a if errMarshal != nil { return errMarshal } + req, errNewRequest := http.NewRequestWithContext( ctx, http.MethodPost, @@ -81,6 +82,7 @@ func (t *HTTPTransport) Call(ctx context.Context, route handler.Route, request a if errNewRequest != nil { return errNewRequest } + httpResponse, errDo := t.httpClient.Do(req) // #nosec G704 -- The client transport must call the caller-configured contentserver endpoint. if errDo != nil { return errDo @@ -90,13 +92,16 @@ func (t *HTTPTransport) Call(ctx context.Context, route handler.Route, request a if httpResponse.StatusCode != http.StatusOK { return errors.New("non 200 reply") } + if httpResponse.Body == nil { return errors.New("empty response body") } + responseBytes, errRead := io.ReadAll(httpResponse.Body) if errRead != nil { return errRead } + return json.Unmarshal(responseBytes, response) } diff --git a/client/httptransport_test.go b/client/httptransport_test.go index eb9b83f..8134972 100644 --- a/client/httptransport_test.go +++ b/client/httptransport_test.go @@ -52,13 +52,16 @@ type GetContentClient interface { func newHTTPClient(tb testing.TB, server *httptest.Server) *client.Client { tb.Helper() + c, err := client.NewHTTPClient(server.URL + pathContentserver) require.NoError(tb, err) + return c } func initHTTPRepoServer(tb testing.TB, l *zap.Logger) *httptest.Server { tb.Helper() r := initRepo(tb, l) + return httptest.NewServer(handler.NewHTTP(l, r)) } diff --git a/client/sockettransport.go b/client/sockettransport.go index 8107768..882a4d2 100644 --- a/client/sockettransport.go +++ b/client/sockettransport.go @@ -40,16 +40,20 @@ func (t *SocketTransport) Call(ctx context.Context, route handler.Route, request if t.connPool.chanDrainPool == nil { return errors.New("connection pool has been drained, client is dead") } + jsonBytes, err := json.Marshal(request) if err != nil { return fmt.Errorf("could not marshal request : %w", err) } + netChan := make(chan net.Conn) t.connPool.chanConnGet <- netChan + conn := <-netChan if conn == nil { return errors.New("could not get a connection") } + returnConn := func(err error) { t.connPool.chanConnReturn <- connReturn{ conn: conn, @@ -70,6 +74,7 @@ func (t *SocketTransport) Call(ctx context.Context, route handler.Route, request returnConn(err) return fmt.Errorf("failed to send request: %w", err) } + written += n } @@ -85,9 +90,11 @@ func (t *SocketTransport) Call(ctx context.Context, route handler.Route, request returnConn(err) return fmt.Errorf("an error occurred while reading the response: %w", err) } + if n == 0 { break } + responseBytes = append(responseBytes, buf[0:n]...) if responseLength == 0 { for index, byte := range responseBytes { @@ -98,11 +105,14 @@ func (t *SocketTransport) Call(ctx context.Context, route handler.Route, request returnConn(err) return errors.New("could not read response length: " + err.Error()) } + responseBytes = responseBytes[index:] + break } } } + if responseLength > 0 && len(responseBytes) == responseLength { break } @@ -120,9 +130,12 @@ func (t *SocketTransport) Call(ctx context.Context, route handler.Route, request returnConn(remoteErrJSONErr) return remoteErr } + return fmt.Errorf("could not unmarshal response : %w %q", remoteErrJSONErr, string(responseBytes)) } + returnConn(nil) + return nil } diff --git a/client/sockettransport_test.go b/client/sockettransport_test.go index 1c88456..6cd17de 100644 --- a/client/sockettransport_test.go +++ b/client/sockettransport_test.go @@ -18,9 +18,11 @@ import ( func BenchmarkSocketClientAndServerGetContent(b *testing.B) { l := zaptest.NewLogger(b) socketServer := initSocketRepoServer(b, l) + socketClient := newSocketClient(b, socketServer.Addr().String()) defer socketClient.Close() defer socketServer.Close() + benchmarkServerAndClientGetContent(b, 30, 100, socketClient) } diff --git a/cmd/http.go b/cmd/http.go index 1e9e09a..e34011b 100644 --- a/cmd/http.go +++ b/cmd/http.go @@ -34,6 +34,7 @@ func NewHTTPCommand() *cobra.Command { } else { comps = cobra.AppendActiveHelp(comps, "This command does not take any more arguments") } + return comps, cobra.ShellCompDirectiveNoFileComp }, RunE: func(cmd *cobra.Command, args []string) error { @@ -80,6 +81,7 @@ func NewHTTPCommand() *cobra.Command { if !r.Loaded() { return errors.New("repo not loaded yet") } + return nil }) // start initial update and handle error @@ -104,6 +106,7 @@ func NewHTTPCommand() *cobra.Command { ) svr.Run() + return nil }, } @@ -154,18 +157,22 @@ func createStorage(ctx context.Context, v *viper.Viper, l *zap.Logger) (repo.Sto if blobBucket == "" { return nil, fmt.Errorf("blob bucket URL is required when storage-type is 'blob' (supported schemes: gs://, s3://, azblob://)") } + if !isValidBlobScheme(blobBucket) { return nil, fmt.Errorf("unsupported blob storage URL scheme in %q; supported schemes: gs://, s3://, azblob://", blobBucket) } + l.Info("using blob storage", zap.String("bucket", blobBucket), zap.String("prefix", blobPrefix), zap.String("provider", detectBlobProvider(blobBucket)), ) + return repo.NewBlobStorage(ctx, blobBucket, blobPrefix) case "filesystem", "": dir := historyDirFlag(v) l.Info("using filesystem storage", zap.String("dir", dir)) + return repo.NewFilesystemStorage(dir) default: return nil, fmt.Errorf("unknown storage type: %s (supported: filesystem, blob)", storageType) @@ -179,6 +186,7 @@ func isValidBlobScheme(bucketURL string) bool { return true } } + return false } diff --git a/cmd/root.go b/cmd/root.go index ced7ae4..aed926a 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -53,5 +53,6 @@ func initConfig() { func newViper() *viper.Viper { v := viper.New() v.AutomaticEnv() + return v } diff --git a/cmd/socket.go b/cmd/socket.go index fe2803f..a3e6262 100644 --- a/cmd/socket.go +++ b/cmd/socket.go @@ -27,6 +27,7 @@ func NewSocketCommand() *cobra.Command { } else { comps = cobra.AppendActiveHelp(comps, "This command does not take any more arguments") } + return comps, cobra.ShellCompDirectiveNoFileComp }, RunE: func(cmd *cobra.Command, args []string) error { @@ -70,6 +71,7 @@ func NewSocketCommand() *cobra.Command { // listen on socket var lc net.ListenConfig + ln, err := lc.Listen(cmd.Context(), "tcp", addressFlag(v)) if err != nil { return err @@ -77,10 +79,12 @@ func NewSocketCommand() *cobra.Command { // start repo up := make(chan bool, 1) + r.OnLoaded(func() { up <- true }) go r.Start(context.Background()) //nolint:errcheck + <-up l.Info("started listening", zap.String("address", addressFlag(v))) @@ -97,6 +101,7 @@ func NewSocketCommand() *cobra.Command { go func() { l.Debug("accepted connection", zap.String("source", conn.RemoteAddr().String())) handle.Serve(conn) + if err := conn.Close(); err != nil { l.Warn("failed to close connection", zap.Error(err)) } diff --git a/cmd/version.go b/cmd/version.go index f3c81ac..eb9feba 100644 --- a/cmd/version.go +++ b/cmd/version.go @@ -17,5 +17,6 @@ func NewVersionCommand() *cobra.Command { fmt.Println(version) }, } + return cmd } diff --git a/content/reponode.go b/content/reponode.go index 2a36377..1e183a7 100644 --- a/content/reponode.go +++ b/content/reponode.go @@ -48,6 +48,7 @@ func (n *RepoNode) InPath(path []*Item) bool { return true } } + return false } @@ -61,6 +62,7 @@ func (n *RepoNode) GetPath(dataFields []string) []*Item { parentNode = parentNode.parent pathLength++ } + parentNode = n.parent var ( @@ -77,6 +79,7 @@ func (n *RepoNode) GetPath(dataFields []string) []*Item { parentNode = parentNode.parent i++ } + return path } @@ -88,6 +91,7 @@ func (n *RepoNode) ToItem(dataFields []string) *Item { item.MimeType = n.MimeType item.Hidden = n.Hidden item.URI = n.URI + item.Groups = n.Groups if dataFields == nil { item.Data = n.Data @@ -98,6 +102,7 @@ func (n *RepoNode) ToItem(dataFields []string) *Item { } } } + return item } @@ -117,6 +122,7 @@ func (n *RepoNode) IsOneOfTheseMimeTypes(mimeTypes []string) bool { if len(mimeTypes) == 0 { return true } + return slices.Contains(mimeTypes, n.MimeType) } @@ -133,6 +139,7 @@ func (n *RepoNode) CanBeAccessedByGroups(groups []string) bool { return true } } + return false } @@ -140,6 +147,7 @@ func (n *RepoNode) CanBeAccessedByGroups(groups []string) bool { func (n *RepoNode) PrintNode(id string, level int) { prefix := strings.Repeat(Indent, level) fmt.Printf("%s %s %s:\n", prefix, id, n.Name) + for key, childNode := range n.Nodes { childNode.PrintNode(key, level+1) } diff --git a/pkg/handler/http.go b/pkg/handler/http.go index c8d6bba..6b98aa7 100644 --- a/pkg/handler/http.go +++ b/pkg/handler/http.go @@ -63,6 +63,7 @@ func (h *HTTP) ServeHTTP(w http.ResponseWriter, r *http.Request) { httputils.ServerError(h.l, w, r, http.StatusMethodNotAllowed, errors.New("method not allowed")) return } + if r.Body == nil { httputils.BadRequestServerError(h.l, w, r, errors.New("empty request body")) return @@ -77,10 +78,12 @@ func (h *HTTP) ServeHTTP(w http.ResponseWriter, r *http.Request) { route := Route(strings.TrimPrefix(r.URL.Path, h.basePath+"/")) if route == RouteGetRepo { w.Header().Set("Content-Type", "application/json") + if err := h.repo.WriteRepoBytes(r.Context(), w); err != nil { h.l.Error("failed to write repo bytes", zap.Error(err)) http.Error(w, "failed to get repo", http.StatusInternalServerError) } + return } @@ -89,6 +92,7 @@ func (h *HTTP) ServeHTTP(w http.ResponseWriter, r *http.Request) { http.Error(w, errReply.Error(), http.StatusInternalServerError) return } + w.Header().Set("Content-Type", "application/json") w.Header().Set("X-Content-Type-Options", "nosniff") // reply is produced by encodeReply -> json.Marshal (jsoniter ConfigCompatibleWithStandardLibrary), @@ -106,6 +110,7 @@ func (h *HTTP) handleRequest(ctx context.Context, r *repo.Repo, route Route, jso start := time.Now() reply, err := h.executeRequest(ctx, r, route, jsonBytes, source) + result := "success" if err != nil { result = "error" @@ -127,9 +132,11 @@ func (h *HTTP) executeRequest(ctx context.Context, r *repo.Repo, route Route, js jsonErr = err return } + processingFunc() } ) + metrics.ContentRequestCounter.WithLabelValues(source).Inc() // handle and process @@ -181,5 +188,6 @@ func (h *HTTP) encodeReply(reply any) (bytes []byte, err error) { if err != nil { h.l.Error("could not encode reply", zap.Error(err)) } + return } diff --git a/pkg/handler/socket.go b/pkg/handler/socket.go index 7779070..95705f3 100644 --- a/pkg/handler/socket.go +++ b/pkg/handler/socket.go @@ -82,17 +82,22 @@ func (h *Socket) Serve(conn net.Conn) { handler, jsonLength, headerErr := h.extractHandlerAndJSONLentgh(header) // reset header header = "" + if headerErr != nil { h.l.Error("invalid request could not read header", zap.Error(headerErr)) + encodedErr, encodingErr := h.encodeReply(responses.NewError(4, "invalid header "+headerErr.Error())) if encodingErr == nil { h.writeResponse(conn, encodedErr) } else { h.l.Error("could not respond to invalid request", zap.Error(encodingErr)) } + return } + h.l.Debug("found json", zap.Int("length", jsonLength)) + if jsonLength > 0 { var ( // let us try to read some json @@ -106,14 +111,17 @@ func (h *Socket) Serve(conn net.Conn) { for jsonLengthCurrent < jsonLength { readRound++ + readLength, jsonReadErr := conn.Read(jsonBytes[jsonLengthCurrent:jsonLength]) if jsonReadErr != nil { // @fixme we need to force a read timeout (SetReadDeadline?), if expected jsonLength is lower than really sent bytes (e.g. if client implements protocol wrong) // @todo should we check for io.EOF here h.l.Error("could not read json - giving up with this client connection", zap.Error(jsonReadErr)) metrics.NumSocketsGauge.WithLabelValues(conn.RemoteAddr().String()).Dec() + return } + jsonLengthCurrent += readLength h.l.Debug("read cycle status", zap.Int("jsonLengthCurrent", jsonLengthCurrent), @@ -128,8 +136,10 @@ func (h *Socket) Serve(conn net.Conn) { // note: connection remains open continue } + h.l.Error("can not read empty json") metrics.NumSocketsGauge.WithLabelValues(conn.RemoteAddr().String()).Dec() + return } // adding to header byte by byte @@ -146,10 +156,12 @@ func (h *Socket) extractHandlerAndJSONLentgh(header string) (route Route, jsonLe if len(headerParts) != 2 { return "", 0, errors.New("invalid header") } + jsonLength, err = strconv.Atoi(headerParts[1]) if err != nil { err = fmt.Errorf("could not parse length in header: %q", header) } + return Route(headerParts[0]), jsonLength, err } @@ -161,8 +173,10 @@ func (h *Socket) execute(route Route, jsonBytes []byte) (reply []byte) { if err := h.repo.WriteRepoBytes(context.Background(), &b); err != nil { h.l.Error("failed to write repo bytes", zap.Error(err)) errorReply, _ := h.encodeReply(responses.NewError(5, "failed to get repo: "+err.Error())) + return errorReply } + return b.Bytes() } @@ -170,6 +184,7 @@ func (h *Socket) execute(route Route, jsonBytes []byte) (reply []byte) { if handlingError != nil { h.l.Error("socketServer.execute failed", zap.Error(handlingError)) } + return reply } @@ -177,18 +192,22 @@ func (h *Socket) writeResponse(conn net.Conn, reply []byte) { headerBytes := []byte(strconv.Itoa(len(reply))) reply = append(headerBytes, reply...) h.l.Debug("replying", zap.String("reply", string(reply))) + n, writeError := conn.Write(reply) if writeError != nil { h.l.Error("socketServer.writeResponse: could not write reply", zap.Error(writeError)) return } + if n < len(reply) { h.l.Error("socketServer.writeResponse: write too short", zap.Int("got", n), zap.Int("expected", len(reply)), ) + return } + h.l.Debug("replied. waiting for next request on open connection") } @@ -196,6 +215,7 @@ func (h *Socket) handleRequest(r *repo.Repo, route Route, jsonBytes []byte, sour start := time.Now() reply, err := h.executeRequest(r, route, jsonBytes, source) + result := "success" if err != nil { result = "error" @@ -217,9 +237,11 @@ func (h *Socket) executeRequest(r *repo.Repo, route Route, jsonBytes []byte, sou jsonErr = err return } + processingFunc() } ) + metrics.ContentRequestCounter.WithLabelValues(source).Inc() // handle and process @@ -272,5 +294,6 @@ func (h *Socket) encodeReply(reply any) (replyBytes []byte, err error) { if err != nil { h.l.Error("could not encode reply", zap.Error(err)) } + return } diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 6822a48..cc97f50 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -74,6 +74,7 @@ func newSummaryVec(name, help string, labels ...string) *prometheus.SummaryVec { Help: help, }, labels) prometheus.MustRegister(vec) + return vec } @@ -85,6 +86,7 @@ func newCounterVec(name, help string, labels ...string) *prometheus.CounterVec { Help: help, }, labels) prometheus.MustRegister(vec) + return vec } @@ -96,5 +98,6 @@ func newGaugeVec(name, help string, labels ...string) *prometheus.GaugeVec { Help: help, }, labels) prometheus.MustRegister(vec) + return vec } diff --git a/pkg/repo/history.go b/pkg/repo/history.go index de73abe..58484a0 100644 --- a/pkg/repo/history.go +++ b/pkg/repo/history.go @@ -72,6 +72,7 @@ func NewHistory(l *zap.Logger, opts ...HistoryOption) (*History, error) { if err != nil { return nil, fmt.Errorf("failed to create default filesystem storage: %w", err) } + inst.storage = storage } @@ -113,11 +114,14 @@ func (h *History) Add(ctx context.Context, jsonBytes []byte) error { func (h *History) GetCurrent(ctx context.Context, buf *bytes.Buffer) error { h.mu.RLock() defer h.mu.RUnlock() + data, err := h.storage.Read(ctx, CurrentKey) if err != nil { return err } + _, err = buf.Write(data) + return err } @@ -125,9 +129,11 @@ func (h *History) GetCurrent(ctx context.Context, buf *bytes.Buffer) error { func (h *History) Close() error { h.mu.Lock() defer h.mu.Unlock() + if h.storage != nil { return h.storage.Close() } + return nil } @@ -148,6 +154,7 @@ func (h *History) getHistory(ctx context.Context) (files []string, err error) { files = append(files, key) } } + return files, nil } @@ -159,6 +166,7 @@ func (h *History) cleanup(ctx context.Context) error { for _, f := range files { h.l.Debug("removing outdated backup", zap.String("file", f)) + if err := h.storage.Delete(ctx, f); err != nil { return fmt.Errorf("could not remove file %s: %w", f, err) } @@ -178,5 +186,6 @@ func (h *History) getFilesForCleanup(ctx context.Context, historyVersions int) ( files = append(files, contentFiles[i]) } } + return files, nil } diff --git a/pkg/repo/history_test.go b/pkg/repo/history_test.go index 2469776..822ddf9 100644 --- a/pkg/repo/history_test.go +++ b/pkg/repo/history_test.go @@ -21,10 +21,12 @@ func TestHistoryCurrent(t *testing.T) { test = []byte("test") b bytes.Buffer ) + err := h.Add(ctx, test) require.NoError(t, err) err = h.GetCurrent(ctx, &b) require.NoError(t, err) + if !bytes.Equal(b.Bytes(), test) { t.Fatalf("expected %q, got %q", string(test), b.String()) } @@ -32,12 +34,14 @@ func TestHistoryCurrent(t *testing.T) { func TestHistoryCleanup(t *testing.T) { ctx := context.Background() + h := testHistory(t) for i := range 50 { err := h.Add(ctx, fmt.Append(nil, i)) require.NoError(t, err) time.Sleep(time.Millisecond * 5) } + err := h.cleanup(ctx) require.NoError(t, err) files, err := h.getHistory(ctx) @@ -84,6 +88,7 @@ func TestHistoryWithStorage(t *testing.T) { require.NoError(t, err) var buf bytes.Buffer + err = h.GetCurrent(ctx, &buf) require.NoError(t, err) assert.Equal(t, "test-data", buf.String()) @@ -98,6 +103,7 @@ func TestHistoryWithBlobStorage(t *testing.T) { ctx := context.Background() bucket, err := blob.OpenBucket(ctx, "mem://") require.NoError(t, err) + defer bucket.Close() storage := NewBlobStorageFromBucket(bucket, "test-prefix") @@ -111,6 +117,7 @@ func TestHistoryWithBlobStorage(t *testing.T) { // Test GetCurrent var buf bytes.Buffer + err = h.GetCurrent(ctx, &buf) require.NoError(t, err) assert.Equal(t, "test-data", buf.String()) @@ -118,6 +125,7 @@ func TestHistoryWithBlobStorage(t *testing.T) { // Test cleanup - add more entries for i := range 5 { time.Sleep(time.Millisecond * 5) // Ensure unique timestamps + err = h.Add(ctx, fmt.Appendf(nil, "data-%d", i)) require.NoError(t, err) } @@ -139,6 +147,7 @@ func testHistory(t *testing.T) *History { l := zaptest.NewLogger(t) h, err := NewHistory(l, HistoryWithHistoryLimit(2), HistoryWithHistoryDir(t.TempDir())) require.NoError(t, err) + return h } @@ -150,5 +159,6 @@ func testHistoryWithTestdata(t *testing.T) *History { require.NoError(t, err) h, err := NewHistory(l, HistoryWithStorage(storage), HistoryWithHistoryLimit(2)) require.NoError(t, err) + return h } diff --git a/pkg/repo/loader.go b/pkg/repo/loader.go index 43b630e..fc7be14 100644 --- a/pkg/repo/loader.go +++ b/pkg/repo/loader.go @@ -32,6 +32,7 @@ type updateResponse struct { func (r *Repo) PollRoutine(ctx context.Context) error { l := r.l.Named("routine.poll") ticker := time.NewTicker(r.pollInterval) + for { select { case <-ctx.Done(): @@ -40,6 +41,7 @@ func (r *Repo) PollRoutine(ctx context.Context) error { case <-ticker.C: chanReponse := make(chan updateResponse) r.updateInProgressChannel <- chanReponse + response := <-chanReponse if response.err == nil { l.Info("update success", zap.String("revision", r.version)) @@ -52,6 +54,7 @@ func (r *Repo) PollRoutine(ctx context.Context) error { func (r *Repo) UpdateRoutine(ctx context.Context) error { l := r.l.Named("routine.update") + for { select { case <-ctx.Done(): @@ -71,12 +74,14 @@ func (r *Repo) UpdateRoutine(ctx context.Context) error { if !r.Loaded() { r.loaded.Store(true) l.Info("initial update success") + if r.onLoaded != nil { r.onLoaded() } } else { l.Info("update success") } + metrics.UpdatesCompletedCounter.WithLabelValues().Inc() } @@ -92,6 +97,7 @@ func (r *Repo) UpdateRoutine(ctx context.Context) error { func (r *Repo) DimensionUpdateRoutine(ctx context.Context) error { l := r.l.Named("routine.dimensionUpdate") + for { select { case <-ctx.Done(): @@ -101,10 +107,13 @@ func (r *Repo) DimensionUpdateRoutine(ctx context.Context) error { l.Debug("received a new dimension", zap.String("dimension", newDimension.Dimension)) err := r._updateDimension(newDimension.Dimension, newDimension.Node) + l.Info("received result") + if err != nil { l.Debug("update failed", zap.Error(err)) } + r.dimensionUpdateDoneChannel <- err } } @@ -112,11 +121,14 @@ func (r *Repo) DimensionUpdateRoutine(ctx context.Context) error { func (r *Repo) updateDimension(dimension string, node *content.RepoNode) error { r.l.Debug("trying to push dimension into update channel", zap.String("dimension", dimension), zap.String("nodeName", node.Name)) + r.dimensionUpdateChannel <- &RepoDimension{ Dimension: dimension, Node: node, } + r.l.Debug("waiting for done signal") + return <-r.dimensionUpdateDoneChannel } @@ -132,6 +144,7 @@ func (r *Repo) _updateDimension(dimension string, newNode *content.RepoNode) err if err != nil { return errors.New("update dimension \"" + dimension + "\" failed when building its directory:: " + err.Error()) } + err = wireAliases(newDirectory) if err != nil { return err @@ -142,6 +155,7 @@ func (r *Repo) _updateDimension(dimension string, newNode *content.RepoNode) err // copy old datastructure to prevent concurrent map access // collect other dimension in the Directory newRepoDirectory := map[string]*Dimension{} + for d, D := range r.Directory() { if d != dimension { newRepoDirectory[d] = D @@ -175,11 +189,13 @@ func buildDirectory(dirNode *content.RepoNode, directory map[string]*content.Rep if ok { return errors.New("duplicate node with id:" + existingNode.ID) } + directory[dirNode.ID] = dirNode // todo handle duplicate uris if _, thereIsAnExistingURINode := uRIDirectory[dirNode.URI]; thereIsAnExistingURINode { return errors.New("duplicate uri: " + dirNode.URI + " (bad node id: " + dirNode.ID + ")") } + uRIDirectory[dirNode.URI] = dirNode for _, childNode := range dirNode.Nodes { err := buildDirectory(childNode, directory, uRIDirectory) @@ -187,6 +203,7 @@ func buildDirectory(dirNode *content.RepoNode, directory map[string]*content.Rep return err } } + return nil } @@ -200,26 +217,32 @@ func wireAliases(directory map[string]*content.RepoNode) error { } } } + return nil } func (r *Repo) loadNodesFromJSON() (nodes map[string]*content.RepoNode, err error) { nodes = make(map[string]*content.RepoNode) + err = json.Unmarshal(r.JSONBufferBytes(), &nodes) if err != nil { r.l.Error("Failed to deserialize nodes", zap.Error(err)) return nil, errors.New("failed to deserialize nodes") } + return nodes, nil } func (r *Repo) tryToRestoreCurrent(ctx context.Context) error { buffer := &bytes.Buffer{} + err := r.history.GetCurrent(ctx, buffer) if err != nil { return err } + r.SetJSONBuffer(buffer) + return r.loadJSONBytes(ctx) } @@ -228,6 +251,7 @@ func (r *Repo) get(ctx context.Context, url string) error { if err != nil { return errors.Wrap(err, "failed to create get repo request") } + response, err := r.httpClient.Do(req) // #nosec G704 -- The repository URL is explicitly configured for this loader. if err != nil { return errors.Wrap(err, "failed to get repo") @@ -246,6 +270,7 @@ func (r *Repo) get(ctx context.Context, url string) error { if err != nil { return errors.Wrap(err, "failed to copy IO stream") } + r.SetJSONBuffer(buffer) return nil @@ -262,6 +287,7 @@ func (r *Repo) update(ctx context.Context) (repoRuntime int64, err error) { // next poll's If-None-Match would elicit a 304 that silently returns // success, masking the staleness until upstream changes content. var newVersion string + if r.poll { req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.url, nil) if err != nil { @@ -275,6 +301,7 @@ func (r *Repo) update(ctx context.Context) (repoRuntime int64, err error) { if strings.HasPrefix(r.version, `"`) || strings.HasPrefix(r.version, `W/"`) { req.Header.Set("If-None-Match", r.version) } + resp, err := r.httpClient.Do(req) // #nosec G704 -- Poll mode intentionally calls the configured poll endpoint. if err != nil { return repoRuntime, err @@ -297,6 +324,7 @@ func (r *Repo) update(ctx context.Context) (repoRuntime int64, err error) { if err != nil { return repoRuntime, errors.New("could not poll latest repo download url, could not read body") } + repoURL = string(responseBytes) // version = ETag if the server sent one, else the URL body. Commit to @@ -305,31 +333,38 @@ func (r *Repo) update(ctx context.Context) (repoRuntime int64, err error) { if newVersion == "" { newVersion = repoURL } + if newVersion == r.version { r.l.Info("repo is up to date", zap.String("version", r.version)) return repoRuntime, nil } + r.l.Info("new repo version", zap.String("version", newVersion)) } err = r.get(ctx, repoURL) repoRuntime = time.Now().UnixNano() - startTimeRepo + if err != nil { // we have no json to load - the repo server did not reply r.l.Debug("failed to load json", zap.Error(err)) return repoRuntime, err } + r.l.Debug("loading json", zap.String("server", repoURL), zap.Int("length", len(r.JSONBufferBytes()))) + nodes, err := r.loadNodesFromJSON() if err != nil { // could not load nodes from json return repoRuntime, err } + err = r.loadNodes(nodes) if err != nil { // repo failed to load nodes return repoRuntime, err } + if r.poll { r.version = newVersion } @@ -360,7 +395,9 @@ func (r *Repo) tryUpdate() (repoRuntime int64, err error) { select { case r.updateInProgressChannel <- c: r.l.Debug("update request added to queue") + ur := <-c + return ur.repoRuntime, ur.err default: r.l.Info("update request accepted, will be processed after the previous update") @@ -379,6 +416,7 @@ func (r *Repo) loadJSONBytes(ctx context.Context) error { zap.String("jsonStart", string(data[len(data)-10:])), ) } + return err } @@ -392,35 +430,44 @@ func (r *Repo) loadJSONBytes(ctx context.Context) error { r.l.Info("added valid JSON to history") } } + return err } func (r *Repo) loadNodes(newNodes map[string]*content.RepoNode) error { var err error + newDimensions := make([]string, 0, len(newNodes)) for dimension, newNode := range newNodes { newDimensions = append(newDimensions, dimension) r.l.Debug("loading nodes for dimension", zap.String("dimension", dimension)) + errLoad := r.updateDimension(dimension, newNode) if errLoad != nil { err = multierr.Append(err, errLoad) } } + if err != nil { return errors.Wrap(err, "failed to update dimension") } + dimensionIsValid := func(dimension string) bool { return slices.Contains(newDimensions, dimension) } // we need to throw away orphaned dimensions directory := map[string]*Dimension{} + for dimension, value := range r.Directory() { if !dimensionIsValid(dimension) { r.l.Info("removing orphaned dimension", zap.String("dimension", dimension)) continue } + directory[dimension] = value } + r.SetDirectory(directory) + return nil } diff --git a/pkg/repo/loader_test.go b/pkg/repo/loader_test.go index 81f3c03..45581c9 100644 --- a/pkg/repo/loader_test.go +++ b/pkg/repo/loader_test.go @@ -28,12 +28,12 @@ func newMinimalRepo(t *testing.T, url string) *Repo { l := zaptest.NewLogger(t) h, err := NewHistory(l, HistoryWithHistoryLimit(2), HistoryWithHistoryDir(t.TempDir())) require.NoError(t, err) + return New(l, url, h, WithPoll(true)) } func TestPollRoutineLogsSuccessfulVersion(t *testing.T) { // t.Parallel() - core, logs := observer.New(zap.InfoLevel) r := New(zap.New(core), "http://example.test/repo", nil, WithPoll(true), WithPollInterval(time.Millisecond)) r.version = `"v1"` @@ -43,13 +43,16 @@ func TestPollRoutineLogsSuccessfulVersion(t *testing.T) { handledUpdate := make(chan struct{}, 1) stopResponder := make(chan struct{}) + responderDone := make(chan struct{}) go func() { defer close(responderDone) + for { select { case resChan := <-r.updateInProgressChannel: resChan <- updateResponse{} + select { case handledUpdate <- struct{}{}: default: @@ -72,6 +75,7 @@ func TestPollRoutineLogsSuccessfulVersion(t *testing.T) { } cancel() + select { case err := <-pollDone: require.NoError(t, err) @@ -111,6 +115,7 @@ func TestUpdate_NoETag_BackwardCompat(t *testing.T) { _, _ = w.Write([]byte("http://" + r.Host + testRepoPath)) //nolint:gosec // r.Host is the test server's local address, not user input case testRepoPath: repoCallCount++ + w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte(testRepoBody)) @@ -125,6 +130,7 @@ func TestUpdate_NoETag_BackwardCompat(t *testing.T) { // Start the background channel-routing goroutines that update() requires. ctx, cancel := context.WithCancel(t.Context()) defer cancel() + go r.UpdateRoutine(ctx) //nolint:errcheck go r.DimensionUpdateRoutine(ctx) //nolint:errcheck @@ -149,8 +155,8 @@ func TestUpdate_NoETag_BackwardCompat(t *testing.T) { // replies 304, causing the loader to skip the body read. func TestUpdate_ETagSetThenNotModified(t *testing.T) { // t.Parallel() - const etagV1 = `"v1"` + var ( pollCallCount int receivedIfNoneMatch string @@ -160,6 +166,7 @@ func TestUpdate_ETagSetThenNotModified(t *testing.T) { switch r.URL.Path { case testPollPath: pollCallCount++ + inm := r.Header.Get("If-None-Match") if inm != "" { receivedIfNoneMatch = inm @@ -169,6 +176,7 @@ func TestUpdate_ETagSetThenNotModified(t *testing.T) { // Confirm the ETag — content unchanged. w.Header().Set("ETag", etagV1) w.WriteHeader(http.StatusNotModified) + return } @@ -193,6 +201,7 @@ func TestUpdate_ETagSetThenNotModified(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() + go r.UpdateRoutine(ctx) //nolint:errcheck go r.DimensionUpdateRoutine(ctx) //nolint:errcheck @@ -214,11 +223,11 @@ func TestUpdate_ETagSetThenNotModified(t *testing.T) { // 200 response the loader updates version and fetches the new repo content. func TestUpdate_ETagChange(t *testing.T) { // t.Parallel() - const ( etagV1 = `"v1"` etagV2 = `"v2"` ) + callCount := 0 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -254,6 +263,7 @@ func TestUpdate_ETagChange(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() + go r.UpdateRoutine(ctx) //nolint:errcheck go r.DimensionUpdateRoutine(ctx) //nolint:errcheck @@ -274,8 +284,8 @@ func TestUpdate_ETagChange(t *testing.T) { // captured ETag — so the next attempt can still send a valid If-None-Match. func TestUpdate_NonOKNon304_ReturnsErrorAndPreservesETag(t *testing.T) { // t.Parallel() - const etagV1 = `"v1"` + callCount := 0 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -307,6 +317,7 @@ func TestUpdate_NonOKNon304_ReturnsErrorAndPreservesETag(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() + go r.UpdateRoutine(ctx) //nolint:errcheck go r.DimensionUpdateRoutine(ctx) //nolint:errcheck @@ -329,8 +340,8 @@ func TestUpdate_NonOKNon304_ReturnsErrorAndPreservesETag(t *testing.T) { // else URL" rule for the version field. func TestUpdate_ETagThenAbsent_FallsBackToURL(t *testing.T) { // t.Parallel() - const etagV1 = `"v1"` + callCount := 0 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -365,6 +376,7 @@ func TestUpdate_ETagThenAbsent_FallsBackToURL(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() + go r.UpdateRoutine(ctx) //nolint:errcheck go r.DimensionUpdateRoutine(ctx) //nolint:errcheck @@ -388,8 +400,8 @@ func TestUpdate_ETagThenAbsent_FallsBackToURL(t *testing.T) { // recovering. This is a regression test for that ordering bug. func TestUpdate_VersionNotCommittedOnLoadFailure(t *testing.T) { // t.Parallel() - const etagV1 = `"v1"` + var pollCallCount int srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -438,7 +450,6 @@ func TestUpdate_VersionNotCommittedOnLoadFailure(t *testing.T) { // startup). func TestUpdate_NoIfNoneMatchOnFirstCall(t *testing.T) { // t.Parallel() - var firstRequestHadIfNoneMatch bool srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -447,6 +458,7 @@ func TestUpdate_NoIfNoneMatchOnFirstCall(t *testing.T) { if r.Header.Get("If-None-Match") != "" { firstRequestHadIfNoneMatch = true } + w.Header().Set("Content-Type", "text/plain") w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte("http://" + r.Host + testRepoPath)) //nolint:gosec // r.Host is the test server's local address, not user input @@ -464,6 +476,7 @@ func TestUpdate_NoIfNoneMatchOnFirstCall(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() + go r.UpdateRoutine(ctx) //nolint:errcheck go r.DimensionUpdateRoutine(ctx) //nolint:errcheck diff --git a/pkg/repo/mock/mock.go b/pkg/repo/mock/mock.go index ef1eb44..7c7297a 100644 --- a/pkg/repo/mock/mock.go +++ b/pkg/repo/mock/mock.go @@ -19,6 +19,7 @@ const ( // GetMockData mock data to run a repo func GetMockData(tb testing.TB) (*httptest.Server, string) { tb.Helper() + _, filename, _, _ := runtime.Caller(0) mockDir := path.Dir(filename) fileServer := http.FileServer(http.Dir(mockDir)) @@ -65,6 +66,7 @@ func MakeValidURIsRequest() *requests.URIs { // MakeValidContentRequest a mock content request func MakeValidContentRequest() *requests.Content { dimensions := []string{dimensionFoo} + return &requests.Content{ URI: "/a", Env: &requests.Env{ diff --git a/pkg/repo/repo.go b/pkg/repo/repo.go index 342cac0..86a843c 100644 --- a/pkg/repo/repo.go +++ b/pkg/repo/repo.go @@ -109,24 +109,28 @@ func (r *Repo) Loaded() bool { func (r *Repo) Directory() map[string]*Dimension { r.directoryLock.RLock() defer r.directoryLock.RUnlock() + return r.directory } func (r *Repo) SetDirectory(v map[string]*Dimension) { r.directoryLock.Lock() defer r.directoryLock.Unlock() + r.directory = v } func (r *Repo) JSONBufferBytes() []byte { r.jsonBufferLock.RLock() defer r.jsonBufferLock.RUnlock() + return r.jsonBuffer.Bytes() } func (r *Repo) SetJSONBuffer(v *bytes.Buffer) { r.jsonBufferLock.Lock() defer r.jsonBufferLock.Unlock() + r.jsonBuffer = v } @@ -144,6 +148,7 @@ func (r *Repo) GetURIs(dimension string, ids []string) map[string]string { for _, id := range ids { uris[id] = r.getURI(dimension, id) } + return uris } @@ -167,18 +172,24 @@ func (r *Repo) GetContent(req *requests.Content) (*content.SiteContent, error) { if err != nil { return nil, errors.Wrap(err, "repo.GetContent invalid request") } + r.l.Debug("repo.GetContent", zap.String("URI", req.URI)) + c := content.NewSiteContent() + resolved, resolvedURI, resolvedDimension, node := r.resolveContent(req.Env.Dimensions, req.URI) if resolved { if !node.CanBeAccessedByGroups(req.Env.Groups) { r.l.Warn("Resolved content cannot be accessed by specified group", zap.String("uri", req.URI)) + c.Status = content.StatusForbidden } else { r.l.Info("Content resolved", zap.String("uri", req.URI)) + c.Status = content.StatusOk c.Data = node.Data } + c.MimeType = node.MimeType c.Dimension = resolvedDimension c.URI = resolvedURI @@ -189,9 +200,11 @@ func (r *Repo) GetContent(req *requests.Content) (*content.SiteContent, error) { for dimensionName := range r.Directory() { uris[dimensionName] = r.getURI(dimensionName, node.ID) } + c.URIs = uris } else { r.l.Info("Content not found", zap.String("URI", req.URI)) + c.Status = content.StatusNotFound c.Dimension = req.Env.Dimensions[0] @@ -209,7 +222,9 @@ func (r *Repo) GetContent(req *requests.Content) (*content.SiteContent, error) { node.Dimension = resolvedDimension } } + c.Nodes = r.getNodes(req.Nodes, req.Env) + return c, nil } @@ -219,6 +234,7 @@ func (r *Repo) GetRepo() map[string]*content.RepoNode { for dimensionName, dimension := range r.Directory() { response[dimensionName] = dimension.Node } + return response } @@ -227,11 +243,13 @@ func (r *Repo) GetRepo() map[string]*content.RepoNode { // The result is wrapped as service response, e.g: {"reply": } func (r *Repo) WriteRepoBytes(ctx context.Context, w io.Writer) error { r.jsonBufferLock.RLock() + var data []byte if r.jsonBuffer != nil && r.jsonBuffer.Len() > 0 { data = make([]byte, r.jsonBuffer.Len()) copy(data, r.jsonBuffer.Bytes()) } + r.jsonBufferLock.RUnlock() if len(data) == 0 { @@ -240,18 +258,22 @@ func (r *Repo) WriteRepoBytes(ctx context.Context, w io.Writer) error { if err := r.history.GetCurrent(ctx, &buf); err != nil { return fmt.Errorf("failed to read repo from storage: %w", err) } + data = buf.Bytes() } if _, err := w.Write([]byte(`{"reply":`)); err != nil { return fmt.Errorf("failed to write repo JSON prefix: %w", err) } + if _, err := w.Write(data); err != nil { return fmt.Errorf("failed to write repo JSON data: %w", err) } + if _, err := w.Write([]byte(`}`)); err != nil { return fmt.Errorf("failed to write repo JSON suffix: %w", err) } + return nil } @@ -304,7 +326,9 @@ func (r *Repo) Update(ctx context.Context) (updateResponse *responses.Update) { updateResponse.Stats.NumberOfURIs += len(dimension.URIDirectory) } } + updateResponse.Stats.OwnRuntime = floatSeconds(time.Since(start).Nanoseconds()) - updateResponse.Stats.RepoRuntime + return updateResponse } @@ -314,9 +338,12 @@ func (r *Repo) Start(ctx context.Context) error { l := r.l.Named("start") up := make(chan bool, 1) + g.Go(func() error { l.Debug("starting update routine") + up <- true + return r.UpdateRoutine(gCtx) }) l.Debug("waiting for UpdateRoutine") @@ -324,13 +351,16 @@ func (r *Repo) Start(ctx context.Context) error { g.Go(func() error { l.Debug("starting dimension update routine") + up <- true + return r.DimensionUpdateRoutine(gCtx) }) l.Debug("waiting for DimensionUpdateRoutine") <-up l.Debug("trying to restore previous repo") + if err := r.tryToRestoreCurrent(ctx); errors.Is(err, os.ErrNotExist) { l.Info("previous repo content file does not exist") } else if err != nil { @@ -348,6 +378,7 @@ func (r *Repo) Start(ctx context.Context) error { if !r.Loaded() { l.Debug("trying to update initial state") + if resp := r.Update(ctx); !resp.Success { l.Error("failed to update initial state", zap.String("error", resp.ErrorMessage), @@ -371,11 +402,13 @@ func (r *Repo) getNodes(nodeRequests map[string]*requests.Node, env *requests.En path []*content.Item nodes = map[string]*content.Node{} ) + for nodeName, nodeRequest := range nodeRequests { if nodeName == "" || nodeRequest.ID == "" { r.l.Warn("invalid node request", zap.Error(errors.New("nodeName or nodeRequest.ID empty"))) continue } + r.l.Debug("adding node", zap.String("name", nodeName), zap.String("requestID", nodeRequest.ID)) groups := env.Groups @@ -388,12 +421,14 @@ func (r *Repo) getNodes(nodeRequests map[string]*requests.Node, env *requests.En if !ok && nodeRequest.Dimension == "" { r.l.Debug("Could not get dimension root node", zap.String("dimension", nodeRequest.Dimension)) + for _, dimension := range env.Dimensions { dimensionNode, ok = r.Directory()[dimension] if ok { r.l.Debug("Found root node in env.Dimensions", zap.String("dimension", dimension)) break } + r.l.Debug("Could NOT find root node in env.Dimensions", zap.String("dimension", dimension)) } } @@ -410,10 +445,13 @@ func (r *Repo) getNodes(nodeRequests map[string]*requests.Node, env *requests.En zap.String("nodeID", nodeRequest.ID), ) metrics.InvalidNodeTreeRequests.WithLabelValues().Inc() + continue } + nodes[nodeName] = r.getNode(treeNode, nodeRequest.Expand, nodeRequest.MimeTypes, path, 0, groups, nodeRequest.DataFields, nodeRequest.ExposeHiddenNodes) } + return nodes } @@ -421,31 +459,38 @@ func (r *Repo) getNodes(nodeRequests map[string]*requests.Node, env *requests.En func (r *Repo) resolveContent(dimensions []string, uri string) (resolved bool, resolvedURI string, resolvedDimension string, repoNode *content.RepoNode) { parts := strings.Split(uri, content.PathSeparator) r.l.Debug("repo.ResolveContent", zap.String("URI", uri)) + for i := len(parts); i > 0; i-- { testURI := strings.Join(parts[0:i], content.PathSeparator) if testURI == "" { testURI = content.PathSeparator } + for _, dimension := range dimensions { if d, ok := r.Directory()[dimension]; ok { r.l.Debug("Checking node", zap.String("dimension", dimension), zap.String("URI", testURI), ) + if repoNode, ok := d.URIDirectory[testURI]; ok { resolved = true + r.l.Debug("Node found", zap.String("URI", testURI), zap.String("destination", repoNode.DestinationID)) + if len(repoNode.DestinationID) > 0 { if destionationNode, destinationNodeOk := d.Directory[repoNode.DestinationID]; destinationNodeOk { repoNode = destionationNode } } + return resolved, testURI, dimension, repoNode } } } } - return + + return resolved, resolvedURI, resolvedDimension, repoNode } func (r *Repo) getURIForNode(dimension string, repoNode *content.RepoNode, recursionLevel int64) (uri string) { @@ -453,14 +498,17 @@ func (r *Repo) getURIForNode(dimension string, repoNode *content.RepoNode, recur uri = repoNode.URI return } + linkedNode, ok := r.Directory()[dimension].Directory[repoNode.LinkID] if ok { if recursionLevel > maxGetURIForNodeRecursionLevel { r.l.Error("maxGetURIForNodeRecursionLevel reached", zap.String("repoNode.ID", repoNode.ID), zap.String("linkID", repoNode.LinkID), zap.String("dimension", dimension)) return "" } + return r.getURIForNode(dimension, linkedNode, recursionLevel+1) } + return } @@ -469,10 +517,12 @@ func (r *Repo) getURI(dimension string, id string) string { if !ok { return "" } + repoNode, ok := directory.Directory[id] if !ok { return "" } + return r.getURIForNode(dimension, repoNode, 0) } @@ -489,6 +539,7 @@ func (r *Repo) getNode( node := content.NewNode() node.Item = repoNode.ToItem(dataFields) r.l.Debug("getNode", zap.String("ID", repoNode.ID)) + for _, childID := range repoNode.Index { childNode := repoNode.Nodes[childID] if (level == 0 || expanded || !expanded && childNode.InPath(path)) && (!childNode.Hidden || exposeHiddenNodes) && childNode.CanBeAccessedByGroups(groups) && childNode.IsOneOfTheseMimeTypes(mimeTypes) { @@ -496,6 +547,7 @@ func (r *Repo) getNode( node.Index = append(node.Index, childID) } } + return node } @@ -503,21 +555,26 @@ func (r *Repo) validateContentRequest(req *requests.Content) (err error) { if req == nil { return errors.New("request must not be nil") } + if len(req.URI) == 0 { return errors.New("request URI must not be empty") } + if req.Env == nil { return errors.New("request.Env must not be nil") } + if len(req.Env.Dimensions) == 0 { return errors.New("request.Env.Dimensions must not be empty") } + for _, envDimension := range req.Env.Dimensions { if !r.hasDimension(envDimension) { availableDimensions := make([]string, 0, len(r.Directory())) for availableDimension := range r.Directory() { availableDimensions = append(availableDimensions, availableDimension) } + return errors.New(fmt.Sprint( "unknown dimension ", envDimension, " in r.Env must be one of ", availableDimensions, @@ -525,6 +582,7 @@ func (r *Repo) validateContentRequest(req *requests.Content) (err error) { )) } } + return nil } diff --git a/pkg/repo/repo_test.go b/pkg/repo/repo_test.go index 3d17a45..016c907 100644 --- a/pkg/repo/repo_test.go +++ b/pkg/repo/repo_test.go @@ -20,14 +20,18 @@ func NewTestRepo(ctx context.Context, l *zap.Logger, url, varDir string) *Repo { if err != nil { panic(err) } + r := New(l, url, h) go r.Start(ctx) //nolint:errcheck + time.Sleep(100 * time.Millisecond) + return r } func assertRepoIsEmpty(t *testing.T, r *Repo, empty bool) { t.Helper() + if empty { if len(r.Directory()) > 0 { t.Fatal("directory should have been empty, but is not") @@ -82,9 +86,11 @@ func TestLoadRepo(t *testing.T) { if !response.Success { t.Fatal("could not load valid repo") } + if response.Stats.OwnRuntime > response.Stats.RepoRuntime { t.Fatal("how could all take less time, than me alone") } + if response.Stats.RepoRuntime < 0.05 { t.Fatal("the server was too fast") } @@ -105,6 +111,7 @@ func BenchmarkLoadRepo(b *testing.B) { b.ReportAllocs() b.ResetTimer() + for n := 0; n < b.N; n++ { response := r.Update(b.Context()) if len(r.Directory()) == 0 { @@ -272,18 +279,21 @@ func TestWriteRepoBytesRace(t *testing.T) { wg.Add(2) go func() { defer wg.Done() + for { select { case <-ctx.Done(): return default: var buf bytes.Buffer + _ = r.WriteRepoBytes(ctx, &buf) } } }() go func() { defer wg.Done() + for { select { case <-ctx.Done(): @@ -295,5 +305,6 @@ func TestWriteRepoBytesRace(t *testing.T) { } }() } + wg.Wait() } diff --git a/pkg/repo/storage_blob.go b/pkg/repo/storage_blob.go index d58a840..bed86cd 100644 --- a/pkg/repo/storage_blob.go +++ b/pkg/repo/storage_blob.go @@ -40,6 +40,7 @@ func NewBlobStorage(ctx context.Context, bucketURL, prefix string) (*BlobStorage if prefix != "" && !strings.HasSuffix(prefix, "/") { prefix += "/" } + return &BlobStorage{ bucket: bucket, prefix: prefix, @@ -53,6 +54,7 @@ func NewBlobStorageFromBucket(bucket *blob.Bucket, prefix string) *BlobStorage { if prefix != "" && !strings.HasSuffix(prefix, "/") { prefix += "/" } + return &BlobStorage{ bucket: bucket, prefix: prefix, @@ -63,6 +65,7 @@ func (b *BlobStorage) Write(ctx context.Context, key string, data []byte) error if err := b.bucket.WriteAll(ctx, b.fullKey(key), data, nil); err != nil { return fmt.Errorf("failed to write blob %q: %w", key, err) } + return nil } @@ -72,8 +75,10 @@ func (b *BlobStorage) Read(ctx context.Context, key string) ([]byte, error) { if gcerrors.Code(err) == gcerrors.NotFound { return nil, os.ErrNotExist } + return nil, fmt.Errorf("failed to read blob %q: %w", key, err) } + return data, nil } @@ -83,25 +88,32 @@ func (b *BlobStorage) List(ctx context.Context, prefix string) ([]string, error) }) var keys []string + for { obj, err := iter.Next(ctx) if errors.Is(err, io.EOF) { break } + if err != nil { return nil, fmt.Errorf("failed to list blobs with prefix %q: %w", prefix, err) } + key := obj.Key if b.prefix != "" { // Skip keys that don't have our prefix (shouldn't happen, but be safe) if !strings.HasPrefix(key, b.prefix) { continue } + key = strings.TrimPrefix(key, b.prefix) } + keys = append(keys, key) } + sort.Sort(sort.Reverse(sort.StringSlice(keys))) + return keys, nil } @@ -111,8 +123,10 @@ func (b *BlobStorage) Delete(ctx context.Context, key string) error { if gcerrors.Code(err) == gcerrors.NotFound { return nil } + return fmt.Errorf("failed to delete blob %q: %w", key, err) } + return nil } @@ -120,6 +134,7 @@ func (b *BlobStorage) Close() error { if err := b.bucket.Close(); err != nil { return fmt.Errorf("failed to close bucket: %w", err) } + return nil } @@ -127,5 +142,6 @@ func (b *BlobStorage) fullKey(key string) string { if b.prefix == "" { return key } + return b.prefix + key } diff --git a/pkg/repo/storage_blob_test.go b/pkg/repo/storage_blob_test.go index 5efcf51..26fc50a 100644 --- a/pkg/repo/storage_blob_test.go +++ b/pkg/repo/storage_blob_test.go @@ -14,10 +14,12 @@ import ( func newTestBlobStorage(t *testing.T, prefix string) *BlobStorage { t.Helper() + ctx := context.Background() bucket, err := blob.OpenBucket(ctx, "mem://") require.NoError(t, err) t.Cleanup(func() { bucket.Close() }) + return NewBlobStorageFromBucket(bucket, prefix) } @@ -179,6 +181,7 @@ func TestBlobStorage_ConcurrentOperations(t *testing.T) { wg.Add(1) go func(i int) { defer wg.Done() + key := "concurrent-key" data := []byte("data") _ = storage.Write(ctx, key, data) @@ -186,6 +189,7 @@ func TestBlobStorage_ConcurrentOperations(t *testing.T) { _, _ = storage.List(ctx, "concurrent-") }(i) } + wg.Wait() } diff --git a/pkg/repo/storage_fs.go b/pkg/repo/storage_fs.go index c7e4acf..9718951 100644 --- a/pkg/repo/storage_fs.go +++ b/pkg/repo/storage_fs.go @@ -20,6 +20,7 @@ func NewFilesystemStorage(baseDir string) (*FilesystemStorage, error) { if err := os.MkdirAll(baseDir, 0700); err != nil { return nil, err } + return &FilesystemStorage{baseDir: baseDir}, nil } @@ -28,10 +29,12 @@ func (f *FilesystemStorage) Write(_ context.Context, key string, data []byte) er defer f.mu.Unlock() path := filepath.Join(f.baseDir, key) + dir := filepath.Dir(path) if err := os.MkdirAll(dir, 0700); err != nil { return err } + return os.WriteFile(path, data, 0600) } @@ -40,6 +43,7 @@ func (f *FilesystemStorage) Read(_ context.Context, key string) ([]byte, error) defer f.mu.RUnlock() path := filepath.Join(f.baseDir, key) + return os.ReadFile(path) } @@ -56,12 +60,15 @@ func (f *FilesystemStorage) List(_ context.Context, prefix string) ([]string, er } var keys []string + for _, entry := range entries { if !entry.IsDir() && strings.HasPrefix(entry.Name(), prefix) { keys = append(keys, entry.Name()) } } + sort.Sort(sort.Reverse(sort.StringSlice(keys))) + return keys, nil } @@ -70,10 +77,12 @@ func (f *FilesystemStorage) Delete(_ context.Context, key string) error { defer f.mu.Unlock() path := filepath.Join(f.baseDir, key) + err := os.Remove(path) if os.IsNotExist(err) { return nil } + return err } diff --git a/pkg/repo/storage_fs_test.go b/pkg/repo/storage_fs_test.go index 0442f50..7f8c721 100644 --- a/pkg/repo/storage_fs_test.go +++ b/pkg/repo/storage_fs_test.go @@ -131,6 +131,7 @@ func TestFilesystemStorage_ConcurrentOperations(t *testing.T) { wg.Add(1) go func(i int) { defer wg.Done() + key := "concurrent-key" data := []byte("data") _ = storage.Write(ctx, key, data) @@ -138,6 +139,7 @@ func TestFilesystemStorage_ConcurrentOperations(t *testing.T) { _, _ = storage.List(ctx, "concurrent-") }(i) } + wg.Wait() }