From 677495c61987608d4c35f49695b620cfea41ede7 Mon Sep 17 00:00:00 2001 From: Michal Fiedorowicz Date: Thu, 30 Oct 2025 12:38:07 +0100 Subject: [PATCH] feat: default branch support (#422) Signed-off-by: Michal Fiedorowicz --- .../auth/mocks/clientinfodecorator.go | 2 +- diode-server/auth/mocks/clientmanager.go | 2 +- diode-server/auth/mocks/metrics.go | 2 +- .../auth/mocks/tokenownershipprovider.go | 2 +- diode-server/auth/mocks/tokenparser.go | 2 +- diode-server/go.mod | 3 +- diode-server/go.sum | 2 + diode-server/ingester/mocks/metrics.go | 2 +- diode-server/ingester/mocks/streamrouter.go | 2 +- diode-server/netboxdiodeplugin/client.go | 65 +++++ diode-server/netboxdiodeplugin/client_test.go | 266 ++++++++++++++++++ .../netboxdiodeplugin/mocks/netboxapi.go | 60 +++- .../reconciler/ingestion_processor.go | 11 +- .../ingestion_processor_internal_test.go | 3 + .../reconciler/ingestion_processor_test.go | 4 + .../reconciler/mocks/ingestionprocessorops.go | 120 +++++++- diode-server/reconciler/mocks/limits.go | 2 +- diode-server/reconciler/mocks/metrics.go | 2 +- diode-server/reconciler/mocks/redisclient.go | 2 +- diode-server/reconciler/mocks/repository.go | 2 +- diode-server/reconciler/ops.go | 78 ++++- diode-server/reconciler/ops/types.go | 3 +- diode-server/reconciler/ops_test.go | 225 +++++++++++++++ .../telemetry/mocks/metricrecorder.go | 2 +- 24 files changed, 842 insertions(+), 22 deletions(-) diff --git a/diode-server/auth/mocks/clientinfodecorator.go b/diode-server/auth/mocks/clientinfodecorator.go index 84bd5012..3c9386a9 100644 --- a/diode-server/auth/mocks/clientinfodecorator.go +++ b/diode-server/auth/mocks/clientinfodecorator.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package mocks diff --git a/diode-server/auth/mocks/clientmanager.go b/diode-server/auth/mocks/clientmanager.go index edb4ff0d..b64bef51 100644 --- a/diode-server/auth/mocks/clientmanager.go +++ b/diode-server/auth/mocks/clientmanager.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package mocks diff --git a/diode-server/auth/mocks/metrics.go b/diode-server/auth/mocks/metrics.go index 77798559..86930ef0 100644 --- a/diode-server/auth/mocks/metrics.go +++ b/diode-server/auth/mocks/metrics.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package mocks diff --git a/diode-server/auth/mocks/tokenownershipprovider.go b/diode-server/auth/mocks/tokenownershipprovider.go index c3f6ecf4..ccbc4a42 100644 --- a/diode-server/auth/mocks/tokenownershipprovider.go +++ b/diode-server/auth/mocks/tokenownershipprovider.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package mocks diff --git a/diode-server/auth/mocks/tokenparser.go b/diode-server/auth/mocks/tokenparser.go index f9d33549..51560ba8 100644 --- a/diode-server/auth/mocks/tokenparser.go +++ b/diode-server/auth/mocks/tokenparser.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package mocks diff --git a/diode-server/go.mod b/diode-server/go.mod index 4087f49c..64d1ce28 100644 --- a/diode-server/go.mod +++ b/diode-server/go.mod @@ -12,7 +12,9 @@ require ( github.com/golang-jwt/jwt/v5 v5.2.2 github.com/google/uuid v1.6.0 github.com/gosimple/slug v1.15.0 + github.com/gowebpki/jcs v1.0.1 github.com/hashicorp/go-retryablehttp v0.7.7 + github.com/hashicorp/golang-lru/v2 v2.0.7 github.com/jackc/pgx/v5 v5.7.5 github.com/kelseyhightower/envconfig v1.4.0 github.com/oklog/run v1.1.0 @@ -67,7 +69,6 @@ require ( github.com/go-ole/go-ole v1.3.0 // indirect github.com/gogo/protobuf v1.3.2 // indirect github.com/gosimple/unidecode v1.0.1 // indirect - github.com/gowebpki/jcs v1.0.1 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3 // indirect github.com/hashicorp/go-cleanhttp v0.5.2 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect diff --git a/diode-server/go.sum b/diode-server/go.sum index 33806506..e6655ba4 100644 --- a/diode-server/go.sum +++ b/diode-server/go.sum @@ -99,6 +99,8 @@ github.com/hashicorp/go-hclog v1.6.3 h1:Qr2kF+eVWjTiYmU7Y31tYlP1h0q/X3Nl3tPGdaB1 github.com/hashicorp/go-hclog v1.6.3/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M= github.com/hashicorp/go-retryablehttp v0.7.7 h1:C8hUCYzor8PIfXHa4UrZkU4VvK8o9ISHxT2Q8+VepXU= github.com/hashicorp/go-retryablehttp v0.7.7/go.mod h1:pkQpWZeYWskR+D1tR2O5OcBFOxfA7DoAO6xtkuQnHTk= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= diff --git a/diode-server/ingester/mocks/metrics.go b/diode-server/ingester/mocks/metrics.go index f950c858..178ffb01 100644 --- a/diode-server/ingester/mocks/metrics.go +++ b/diode-server/ingester/mocks/metrics.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package mocks diff --git a/diode-server/ingester/mocks/streamrouter.go b/diode-server/ingester/mocks/streamrouter.go index 832839dc..7072b5d1 100644 --- a/diode-server/ingester/mocks/streamrouter.go +++ b/diode-server/ingester/mocks/streamrouter.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package mocks diff --git a/diode-server/netboxdiodeplugin/client.go b/diode-server/netboxdiodeplugin/client.go index 5f511107..292c7e56 100644 --- a/diode-server/netboxdiodeplugin/client.go +++ b/diode-server/netboxdiodeplugin/client.go @@ -56,6 +56,9 @@ const ( // ErrInvalidTimeout is an error for invalid timeout value var ErrInvalidTimeout = errors.New("invalid timeout value") +// ErrDefaultBranchNotFound is returned when the default branch is not found (e.g. no endpoint available in older plugin versions) +var ErrDefaultBranchNotFound = errors.New("default branch not found") + // ChangeSetResult represents a change set result type ChangeSetResult struct { ID string `json:"id"` @@ -263,6 +266,9 @@ type NetBoxAPI interface { // ApplyChangeSet applies a change set ApplyChangeSet(context.Context, ApplyChangeSetRequest) (*ChangeSetResult, error) + + // GetDefaultBranch gets the default branch from NetBox plugin settings + GetDefaultBranch(context.Context) (*Branch, error) } // GenerateDiffRequest represents a generate diff request @@ -412,3 +418,62 @@ func (c *Client) ApplyChangeSet(ctx context.Context, payload ApplyChangeSetReque return &changeSetResult, nil } + +// GetDefaultBranchResponse represents the response from the default-branch endpoint +type GetDefaultBranchResponse struct { + Branch *Branch `json:"branch"` +} + +// GetDefaultBranch gets the default branch from NetBox plugin settings +func (c *Client) GetDefaultBranch(ctx context.Context) (*Branch, error) { + endpointURL, err := url.Parse(fmt.Sprintf("%s/default-branch/", c.baseURL.String())) + if err != nil { + return nil, err + } + + if err := c.limiter.Wait(ctx); err != nil { + return nil, err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpointURL.String(), nil) + if err != nil { + return nil, err + } + + resp, err := c.http.Do(req) + if err != nil { + return nil, err + } + defer func() { + if closeErr := resp.Body.Close(); closeErr != nil { + c.logger.Warn("failed to close response body", "error", closeErr) + } + }() + + respBytes, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("failed to read response body %w", err) + } + + // Log response body only for successful responses to avoid noise from 404s (endpoint not available on older plugin versions) + if resp.StatusCode < http.StatusBadRequest { + c.logger.Debug("get default branch", "statusCode", resp.StatusCode, "response", string(respBytes)) + } else { + c.logger.Debug("get default branch", "statusCode", resp.StatusCode) + } + + if resp.StatusCode >= http.StatusBadRequest { + // Return sentinel error for 404 to allow callers to cache and handle gracefully + if resp.StatusCode == http.StatusNotFound { + return nil, ErrDefaultBranchNotFound + } + return nil, fmt.Errorf("get default branch failed with status %d: %s", resp.StatusCode, string(respBytes)) + } + + var defaultBranchResponse GetDefaultBranchResponse + if err = json.Unmarshal(respBytes, &defaultBranchResponse); err != nil { + return nil, fmt.Errorf("failed to unmarshal response body %w", err) + } + + return defaultBranchResponse.Branch, nil +} diff --git a/diode-server/netboxdiodeplugin/client_test.go b/diode-server/netboxdiodeplugin/client_test.go index 2512ae4c..96839801 100644 --- a/diode-server/netboxdiodeplugin/client_test.go +++ b/diode-server/netboxdiodeplugin/client_test.go @@ -1022,6 +1022,272 @@ func TestApplyChangeSetRateLimiting(t *testing.T) { } } +func TestGetDefaultBranch(t *testing.T) { + tests := []struct { + name string + baseURL string + diodeAuthTokenURL string + diodeToNetBoxClientID string + diodeToNetBoxClientSecret string + mockServerResponse string + mockStatusCode int + rateLimiterRPS int + rateLimiterBurst int + maxRetries int + expectedBranch *netboxdiodeplugin.Branch + shouldError bool + expectedErrorString string + }{ + { + name: "successful branch retrieval", + baseURL: "http://", + diodeAuthTokenURL: "http://diode-auth:8000/diode/auth/token", + diodeToNetBoxClientID: "test", + diodeToNetBoxClientSecret: "test", + mockServerResponse: `{"branch": {"id": "abc123", "name": "dev-branch"}}`, + mockStatusCode: http.StatusOK, + rateLimiterRPS: 1, + rateLimiterBurst: 1, + maxRetries: 3, + expectedBranch: &netboxdiodeplugin.Branch{ + ID: "abc123", + Name: "dev-branch", + }, + shouldError: false, + }, + { + name: "no default branch (null)", + baseURL: "http://", + diodeAuthTokenURL: "http://diode-auth:8000/diode/auth/token", + diodeToNetBoxClientID: "test", + diodeToNetBoxClientSecret: "test", + mockServerResponse: `{"branch": null}`, + mockStatusCode: http.StatusOK, + rateLimiterRPS: 1, + rateLimiterBurst: 1, + maxRetries: 3, + expectedBranch: nil, + shouldError: false, + }, + { + name: "HTTP 500 error", + baseURL: "http://", + diodeAuthTokenURL: "http://diode-auth:8000/diode/auth/token", + diodeToNetBoxClientID: "test", + diodeToNetBoxClientSecret: "test", + mockServerResponse: `{"error": "Internal server error"}`, + mockStatusCode: http.StatusInternalServerError, + rateLimiterRPS: 1, + rateLimiterBurst: 1, + maxRetries: 3, + expectedBranch: nil, + shouldError: true, + expectedErrorString: "get default branch failed with status 500", + }, + { + name: "HTTP 404 error returns sentinel error", + baseURL: "http://", + diodeAuthTokenURL: "http://diode-auth:8000/diode/auth/token", + diodeToNetBoxClientID: "test", + diodeToNetBoxClientSecret: "test", + mockServerResponse: `{"error": "Not found"}`, + mockStatusCode: http.StatusNotFound, + rateLimiterRPS: 1, + rateLimiterBurst: 1, + maxRetries: 3, + expectedBranch: nil, + shouldError: true, + expectedErrorString: "default branch not found", + }, + { + name: "invalid JSON response", + baseURL: "http://", + diodeAuthTokenURL: "http://diode-auth:8000/diode/auth/token", + diodeToNetBoxClientID: "test", + diodeToNetBoxClientSecret: "test", + mockServerResponse: `{invalid json}`, + mockStatusCode: http.StatusOK, + rateLimiterRPS: 1, + rateLimiterBurst: 1, + maxRetries: 3, + expectedBranch: nil, + shouldError: true, + expectedErrorString: "failed to unmarshal response body", + }, + { + name: "HTML error response", + baseURL: "http://", + diodeAuthTokenURL: "http://diode-auth:8000/diode/auth/token", + diodeToNetBoxClientID: "test", + diodeToNetBoxClientSecret: "test", + mockServerResponse: `

500 Internal Server Error

`, + mockStatusCode: http.StatusOK, + rateLimiterRPS: 1, + rateLimiterBurst: 1, + maxRetries: 3, + expectedBranch: nil, + shouldError: true, + expectedErrorString: "failed to unmarshal response body", + }, + } + + logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug, AddSource: false})) + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cleanUpEnvVars() + + expectedToken := "mocked-token" + authTokenURL := "/diode/auth/token" + mockOAuth2Server := newMockOAuth2Server(authTokenURL, requireCredentials(tt.diodeToNetBoxClientID, tt.diodeToNetBoxClientSecret), expectedToken) + defer mockOAuth2Server.Close() + + mockOAuth2ServerURL := mockOAuth2Server.URL + authTokenURL + + handler := func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, http.MethodGet, r.Method) + assert.Equal(t, "/api/diode/default-branch/", r.URL.Path) + assert.Equal(t, fmt.Sprintf("Bearer %s", expectedToken), r.Header.Get("Authorization")) + assert.Equal(t, fmt.Sprintf("%s/%s", netboxdiodeplugin.SDKName, netboxdiodeplugin.SDKVersion), r.Header.Get("User-Agent")) + + w.WriteHeader(tt.mockStatusCode) + _, _ = w.Write([]byte(tt.mockServerResponse)) + } + mux := http.NewServeMux() + mux.HandleFunc("/api/diode/default-branch/", handler) + ts := httptest.NewServer(mux) + defer ts.Close() + + baseURL := fmt.Sprintf("%s/api/diode", ts.URL) + client, err := netboxdiodeplugin.NewClient( + netboxdiodeplugin.ClientOptions{ + Logger: logger, + BaseURL: baseURL, + ClientID: tt.diodeToNetBoxClientID, + ClientSecret: tt.diodeToNetBoxClientSecret, + TokenURL: mockOAuth2ServerURL, + RateLimitRPS: tt.rateLimiterRPS, + RateLimitBurstRPS: tt.rateLimiterBurst, + MaxRetries: tt.maxRetries, + }) + require.NoError(t, err) + + branch, err := client.GetDefaultBranch(context.Background()) + + if tt.shouldError { + require.Error(t, err) + if tt.expectedErrorString != "" { + assert.Contains(t, err.Error(), tt.expectedErrorString) + } + // For 404 errors, verify it's the sentinel error + if tt.mockStatusCode == http.StatusNotFound { + assert.ErrorIs(t, err, netboxdiodeplugin.ErrDefaultBranchNotFound) + } + assert.Nil(t, branch) + return + } + + require.NoError(t, err) + assert.Equal(t, tt.expectedBranch, branch) + }) + } +} + +func TestGetDefaultBranchRateLimiting(t *testing.T) { + tests := []struct { + name string + baseURL string + diodeToNetBoxClientID string + diodeToNetBoxClientSecret string + expectedCalls int + mockServerResponse string + mockStatusCode int + rateLimiterRPS int + rateLimiterBurst int + maxRetries int + expectedBranch *netboxdiodeplugin.Branch + shouldError bool + }{ + { + name: "rate limited requests", + diodeToNetBoxClientID: "test", + diodeToNetBoxClientSecret: "test", + expectedCalls: 2, + mockServerResponse: `{"branch": {"id": "abc123", "name": "dev-branch"}}`, + mockStatusCode: http.StatusOK, + rateLimiterRPS: 1, + rateLimiterBurst: 1, + maxRetries: 3, + expectedBranch: &netboxdiodeplugin.Branch{ + ID: "abc123", + Name: "dev-branch", + }, + shouldError: false, + }, + } + + logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug, AddSource: false})) + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cleanUpEnvVars() + actualCalls := 0 + + expectedToken := "mocked-token" + authTokenURL := "/diode/auth/token" + mockOAuth2Server := newMockOAuth2Server(authTokenURL, requireCredentials(tt.diodeToNetBoxClientID, tt.diodeToNetBoxClientSecret), expectedToken) + defer mockOAuth2Server.Close() + + mockOAuth2ServerURL := mockOAuth2Server.URL + authTokenURL + + handler := func(w http.ResponseWriter, r *http.Request) { + actualCalls++ + assert.Equal(t, http.MethodGet, r.Method) + assert.Equal(t, "/api/diode/default-branch/", r.URL.Path) + assert.Equal(t, fmt.Sprintf("Bearer %s", expectedToken), r.Header.Get("Authorization")) + assert.Equal(t, fmt.Sprintf("%s/%s", netboxdiodeplugin.SDKName, netboxdiodeplugin.SDKVersion), r.Header.Get("User-Agent")) + + w.WriteHeader(tt.mockStatusCode) + _, _ = w.Write([]byte(tt.mockServerResponse)) + } + mux := http.NewServeMux() + mux.HandleFunc("/api/diode/default-branch/", handler) + ts := httptest.NewServer(mux) + defer ts.Close() + + baseURL := fmt.Sprintf("%s/api/diode", ts.URL) + + client, err := netboxdiodeplugin.NewClient( + netboxdiodeplugin.ClientOptions{ + Logger: logger, + BaseURL: baseURL, + ClientID: tt.diodeToNetBoxClientID, + ClientSecret: tt.diodeToNetBoxClientSecret, + TokenURL: mockOAuth2ServerURL, + RateLimitRPS: tt.rateLimiterRPS, + RateLimitBurstRPS: tt.rateLimiterBurst, + MaxRetries: tt.maxRetries, + }) + require.NoError(t, err) + + // Make two calls to test rate limiting + branch, err := client.GetDefaultBranch(context.Background()) + _, _ = client.GetDefaultBranch(context.Background()) + + if tt.shouldError { + require.Error(t, err) + assert.Nil(t, branch) + return + } + + require.NoError(t, err) + assert.Equal(t, tt.expectedBranch, branch) + assert.Equal(t, tt.mockStatusCode, http.StatusOK) + assert.Equal(t, tt.expectedCalls, actualCalls) + }) + } +} + func cleanUpEnvVars() { _ = os.Unsetenv(netboxdiodeplugin.TimeoutSecondsEnvVarName) _ = os.Unsetenv(netboxdiodeplugin.TLSSkipVerifyEnvVarName) diff --git a/diode-server/netboxdiodeplugin/mocks/netboxapi.go b/diode-server/netboxdiodeplugin/mocks/netboxapi.go index 755e5385..af57e14c 100644 --- a/diode-server/netboxdiodeplugin/mocks/netboxapi.go +++ b/diode-server/netboxdiodeplugin/mocks/netboxapi.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package mocks @@ -140,6 +140,64 @@ func (_c *NetBoxAPI_GenerateDiff_Call) RunAndReturn(run func(context.Context, ne return _c } +// GetDefaultBranch provides a mock function with given fields: _a0 +func (_m *NetBoxAPI) GetDefaultBranch(_a0 context.Context) (*netboxdiodeplugin.Branch, error) { + ret := _m.Called(_a0) + + if len(ret) == 0 { + panic("no return value specified for GetDefaultBranch") + } + + var r0 *netboxdiodeplugin.Branch + var r1 error + if rf, ok := ret.Get(0).(func(context.Context) (*netboxdiodeplugin.Branch, error)); ok { + return rf(_a0) + } + if rf, ok := ret.Get(0).(func(context.Context) *netboxdiodeplugin.Branch); ok { + r0 = rf(_a0) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*netboxdiodeplugin.Branch) + } + } + + if rf, ok := ret.Get(1).(func(context.Context) error); ok { + r1 = rf(_a0) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// NetBoxAPI_GetDefaultBranch_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetDefaultBranch' +type NetBoxAPI_GetDefaultBranch_Call struct { + *mock.Call +} + +// GetDefaultBranch is a helper method to define mock.On call +// - _a0 context.Context +func (_e *NetBoxAPI_Expecter) GetDefaultBranch(_a0 interface{}) *NetBoxAPI_GetDefaultBranch_Call { + return &NetBoxAPI_GetDefaultBranch_Call{Call: _e.mock.On("GetDefaultBranch", _a0)} +} + +func (_c *NetBoxAPI_GetDefaultBranch_Call) Run(run func(_a0 context.Context)) *NetBoxAPI_GetDefaultBranch_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context)) + }) + return _c +} + +func (_c *NetBoxAPI_GetDefaultBranch_Call) Return(_a0 *netboxdiodeplugin.Branch, _a1 error) *NetBoxAPI_GetDefaultBranch_Call { + _c.Call.Return(_a0, _a1) + return _c +} + +func (_c *NetBoxAPI_GetDefaultBranch_Call) RunAndReturn(run func(context.Context) (*netboxdiodeplugin.Branch, error)) *NetBoxAPI_GetDefaultBranch_Call { + _c.Call.Return(run) + return _c +} + // NewNetBoxAPI creates a new instance of NetBoxAPI. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. // The first argument is typically a *testing.T value. func NewNetBoxAPI(t interface { diff --git a/diode-server/reconciler/ingestion_processor.go b/diode-server/reconciler/ingestion_processor.go index d46c1268..98d6bf72 100644 --- a/diode-server/reconciler/ingestion_processor.go +++ b/diode-server/reconciler/ingestion_processor.go @@ -21,6 +21,7 @@ import ( "github.com/netboxlabs/diode/diode-server/gen/diode/v1/diodepb" "github.com/netboxlabs/diode/diode-server/gen/diode/v1/reconcilerpb" "github.com/netboxlabs/diode/diode-server/gen/netbox" + "github.com/netboxlabs/diode/diode-server/netboxdiodeplugin" "github.com/netboxlabs/diode/diode-server/reconciler/changeset" "github.com/netboxlabs/diode/diode-server/reconciler/ops" "github.com/netboxlabs/diode/diode-server/sentry" @@ -76,6 +77,7 @@ type IngestionLogToProcess struct { ingestionLog *reconcilerpb.IngestionLog changeSetID int32 changeSet *changeset.ChangeSet + branchID string // the branch ID for this ingestion log (empty string means main branch) } // IngestionProcessorOps represents the basic operations that the ingestion processor performs @@ -83,6 +85,8 @@ type IngestionProcessorOps interface { CreateIngestionLog(ctx context.Context, ingestionLog *reconcilerpb.IngestionLog, sourceMetadata []byte) (*ops.CreateIngestionLogResult, error) GenerateChangeSet(ctx context.Context, ingestionLogID int32, ingestionLog *reconcilerpb.IngestionLog, branchID string) (*int32, *changeset.ChangeSet, error) ApplyChangeSet(ctx context.Context, ingestionLogID int32, ingestionLog *reconcilerpb.IngestionLog, changeSetID int32, changeSet *changeset.ChangeSet) error + DefaultBranch(ctx context.Context) (*netboxdiodeplugin.Branch, error) + RefreshDefaultBranch(ctx context.Context) (*netboxdiodeplugin.Branch, error) } // NewIngestionProcessor creates a new ingestion processor @@ -314,7 +318,7 @@ func (p *IngestionProcessor) GenerateChangeSet(ctx context.Context, generateChan return } - id, changeSet, err := p.ops.GenerateChangeSet(ctx, msg.ingestionLogID, msg.ingestionLog, "") + id, changeSet, err := p.ops.GenerateChangeSet(ctx, msg.ingestionLogID, msg.ingestionLog, msg.branchID) if err != nil { p.logger.Error("error generating changeset", "error", err) p.metrics.RecordChangeSetCreate(ctx, false, 0) @@ -329,6 +333,7 @@ func (p *IngestionProcessor) GenerateChangeSet(ctx context.Context, generateChan ingestionLog: msg.ingestionLog, changeSetID: *id, changeSet: changeSet, + branchID: msg.branchID, } } } @@ -379,6 +384,9 @@ func (p *IngestionProcessor) CreateIngestionLogs(ctx context.Context, ingestReq errs := make([]error, 0) + // Ensure the current default branch is retrieved + _, _ = p.ops.RefreshDefaultBranch(ctx) + for i, v := range ingestReq.GetEntities() { if v.GetEntity() == nil { errs = append(errs, fmt.Errorf("entity at index %d is nil", i)) @@ -437,6 +445,7 @@ func (p *IngestionProcessor) CreateIngestionLogs(ctx context.Context, ingestReq generateIngestionLogChan <- IngestionLogToProcess{ ingestionLogID: id, ingestionLog: ingestionLog, + branchID: result.BranchID, } } diff --git a/diode-server/reconciler/ingestion_processor_internal_test.go b/diode-server/reconciler/ingestion_processor_internal_test.go index d9579442..65ae3c62 100644 --- a/diode-server/reconciler/ingestion_processor_internal_test.go +++ b/diode-server/reconciler/ingestion_processor_internal_test.go @@ -194,6 +194,9 @@ func TestHandleStreamMessage(t *testing.T) { }, } } + // Mock GetDefaultBranch to return nil (no default branch) + mockNbClient.On("GetDefaultBranch", mock.Anything).Return((*netboxdiodeplugin.Branch)(nil), nil) + if tt.reconcilerError { mockNbClient.On("GenerateDiff", mock.Anything, mock.Anything).Return(nil, errors.New("prepare error")) } else { diff --git a/diode-server/reconciler/ingestion_processor_test.go b/diode-server/reconciler/ingestion_processor_test.go index 724fd9be..74403123 100644 --- a/diode-server/reconciler/ingestion_processor_test.go +++ b/diode-server/reconciler/ingestion_processor_test.go @@ -282,6 +282,9 @@ func TestIngestionProcessorStart(t *testing.T) { mockRepository.On("CreateChangeSet", mock.Anything, mock.Anything, mock.Anything).Return(int32Ptr(1), nil) mockRepository.On("TruncateChangeSets", mock.Anything, mock.Anything, mock.Anything).Return(nil) + // Mock GetDefaultBranch to return nil (no default branch) + mockNetBoxClient.On("GetDefaultBranch", mock.Anything).Return((*netboxdiodeplugin.Branch)(nil), nil) + mockNetBoxClient.On("GenerateDiff", mock.Anything, mock.Anything).Return(&netboxdiodeplugin.ChangeSetResult{ ChangeSet: &netboxdiodeplugin.ChangeSet{ ID: "test-changeset-id", @@ -463,6 +466,7 @@ func TestIngestionProcessor_DuplicateHandling(t *testing.T) { WasDuplicate: true, } + mockOps.On("RefreshDefaultBranch", mock.Anything).Return((*netboxdiodeplugin.Branch)(nil), nil) mockOps.On("CreateIngestionLog", mock.Anything, mock.Anything, mock.Anything).Return(duplicateResult, nil) if !tt.expectSkipProcessing { diff --git a/diode-server/reconciler/mocks/ingestionprocessorops.go b/diode-server/reconciler/mocks/ingestionprocessorops.go index e8152985..ea723a7c 100644 --- a/diode-server/reconciler/mocks/ingestionprocessorops.go +++ b/diode-server/reconciler/mocks/ingestionprocessorops.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package mocks @@ -9,6 +9,8 @@ import ( mock "github.com/stretchr/testify/mock" + netboxdiodeplugin "github.com/netboxlabs/diode/diode-server/netboxdiodeplugin" + ops "github.com/netboxlabs/diode/diode-server/reconciler/ops" reconcilerpb "github.com/netboxlabs/diode/diode-server/gen/diode/v1/reconcilerpb" @@ -137,6 +139,64 @@ func (_c *IngestionProcessorOps_CreateIngestionLog_Call) RunAndReturn(run func(c return _c } +// DefaultBranch provides a mock function with given fields: ctx +func (_m *IngestionProcessorOps) DefaultBranch(ctx context.Context) (*netboxdiodeplugin.Branch, error) { + ret := _m.Called(ctx) + + if len(ret) == 0 { + panic("no return value specified for DefaultBranch") + } + + var r0 *netboxdiodeplugin.Branch + var r1 error + if rf, ok := ret.Get(0).(func(context.Context) (*netboxdiodeplugin.Branch, error)); ok { + return rf(ctx) + } + if rf, ok := ret.Get(0).(func(context.Context) *netboxdiodeplugin.Branch); ok { + r0 = rf(ctx) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*netboxdiodeplugin.Branch) + } + } + + if rf, ok := ret.Get(1).(func(context.Context) error); ok { + r1 = rf(ctx) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// IngestionProcessorOps_DefaultBranch_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'DefaultBranch' +type IngestionProcessorOps_DefaultBranch_Call struct { + *mock.Call +} + +// DefaultBranch is a helper method to define mock.On call +// - ctx context.Context +func (_e *IngestionProcessorOps_Expecter) DefaultBranch(ctx interface{}) *IngestionProcessorOps_DefaultBranch_Call { + return &IngestionProcessorOps_DefaultBranch_Call{Call: _e.mock.On("DefaultBranch", ctx)} +} + +func (_c *IngestionProcessorOps_DefaultBranch_Call) Run(run func(ctx context.Context)) *IngestionProcessorOps_DefaultBranch_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context)) + }) + return _c +} + +func (_c *IngestionProcessorOps_DefaultBranch_Call) Return(_a0 *netboxdiodeplugin.Branch, _a1 error) *IngestionProcessorOps_DefaultBranch_Call { + _c.Call.Return(_a0, _a1) + return _c +} + +func (_c *IngestionProcessorOps_DefaultBranch_Call) RunAndReturn(run func(context.Context) (*netboxdiodeplugin.Branch, error)) *IngestionProcessorOps_DefaultBranch_Call { + _c.Call.Return(run) + return _c +} + // GenerateChangeSet provides a mock function with given fields: ctx, ingestionLogID, ingestionLog, branchID func (_m *IngestionProcessorOps) GenerateChangeSet(ctx context.Context, ingestionLogID int32, ingestionLog *reconcilerpb.IngestionLog, branchID string) (*int32, *changeset.ChangeSet, error) { ret := _m.Called(ctx, ingestionLogID, ingestionLog, branchID) @@ -207,6 +267,64 @@ func (_c *IngestionProcessorOps_GenerateChangeSet_Call) RunAndReturn(run func(co return _c } +// RefreshDefaultBranch provides a mock function with given fields: ctx +func (_m *IngestionProcessorOps) RefreshDefaultBranch(ctx context.Context) (*netboxdiodeplugin.Branch, error) { + ret := _m.Called(ctx) + + if len(ret) == 0 { + panic("no return value specified for RefreshDefaultBranch") + } + + var r0 *netboxdiodeplugin.Branch + var r1 error + if rf, ok := ret.Get(0).(func(context.Context) (*netboxdiodeplugin.Branch, error)); ok { + return rf(ctx) + } + if rf, ok := ret.Get(0).(func(context.Context) *netboxdiodeplugin.Branch); ok { + r0 = rf(ctx) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*netboxdiodeplugin.Branch) + } + } + + if rf, ok := ret.Get(1).(func(context.Context) error); ok { + r1 = rf(ctx) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// IngestionProcessorOps_RefreshDefaultBranch_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RefreshDefaultBranch' +type IngestionProcessorOps_RefreshDefaultBranch_Call struct { + *mock.Call +} + +// RefreshDefaultBranch is a helper method to define mock.On call +// - ctx context.Context +func (_e *IngestionProcessorOps_Expecter) RefreshDefaultBranch(ctx interface{}) *IngestionProcessorOps_RefreshDefaultBranch_Call { + return &IngestionProcessorOps_RefreshDefaultBranch_Call{Call: _e.mock.On("RefreshDefaultBranch", ctx)} +} + +func (_c *IngestionProcessorOps_RefreshDefaultBranch_Call) Run(run func(ctx context.Context)) *IngestionProcessorOps_RefreshDefaultBranch_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context)) + }) + return _c +} + +func (_c *IngestionProcessorOps_RefreshDefaultBranch_Call) Return(_a0 *netboxdiodeplugin.Branch, _a1 error) *IngestionProcessorOps_RefreshDefaultBranch_Call { + _c.Call.Return(_a0, _a1) + return _c +} + +func (_c *IngestionProcessorOps_RefreshDefaultBranch_Call) RunAndReturn(run func(context.Context) (*netboxdiodeplugin.Branch, error)) *IngestionProcessorOps_RefreshDefaultBranch_Call { + _c.Call.Return(run) + return _c +} + // NewIngestionProcessorOps creates a new instance of IngestionProcessorOps. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. // The first argument is typically a *testing.T value. func NewIngestionProcessorOps(t interface { diff --git a/diode-server/reconciler/mocks/limits.go b/diode-server/reconciler/mocks/limits.go index 857fe8ae..b546714c 100644 --- a/diode-server/reconciler/mocks/limits.go +++ b/diode-server/reconciler/mocks/limits.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package mocks diff --git a/diode-server/reconciler/mocks/metrics.go b/diode-server/reconciler/mocks/metrics.go index 4663e479..a00289ed 100644 --- a/diode-server/reconciler/mocks/metrics.go +++ b/diode-server/reconciler/mocks/metrics.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package mocks diff --git a/diode-server/reconciler/mocks/redisclient.go b/diode-server/reconciler/mocks/redisclient.go index 9049fb99..d3fc0624 100644 --- a/diode-server/reconciler/mocks/redisclient.go +++ b/diode-server/reconciler/mocks/redisclient.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package mocks diff --git a/diode-server/reconciler/mocks/repository.go b/diode-server/reconciler/mocks/repository.go index 3446ce8f..a47e9fb9 100644 --- a/diode-server/reconciler/mocks/repository.go +++ b/diode-server/reconciler/mocks/repository.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package mocks diff --git a/diode-server/reconciler/ops.go b/diode-server/reconciler/ops.go index 358aac64..44f6c530 100644 --- a/diode-server/reconciler/ops.go +++ b/diode-server/reconciler/ops.go @@ -6,6 +6,9 @@ import ( "errors" "fmt" "log/slog" + "time" + + "github.com/hashicorp/golang-lru/v2/expirable" "github.com/netboxlabs/diode/diode-server/entityhash" diodeErrors "github.com/netboxlabs/diode/diode-server/errors" @@ -18,6 +21,11 @@ import ( "github.com/netboxlabs/diode/diode-server/sentry" ) +const ( + // DefaultBranchCacheTTL is the default TTL for caching the default branch + DefaultBranchCacheTTL = 60 * time.Second +) + // Limits is an interface that provides limits for the reconciler operations to enforce type Limits interface { MaxChangeSetsPerIngestionLog() int32 @@ -37,6 +45,9 @@ type Ops struct { nbClient netboxdiodeplugin.NetBoxAPI logger *slog.Logger limits Limits + + // Cache for default branch (5 minute TTL, size 1 since we only cache one value) + branchCache *expirable.LRU[string, *netboxdiodeplugin.Branch] } // NewOps creates a new Ops @@ -45,14 +56,56 @@ func NewOps(repository Repository, nbClient netboxdiodeplugin.NetBoxAPI, logger limits = &DefaultLimits{} } + // Create LRU cache with size 1 (we only cache one default branch) + branchCache := expirable.NewLRU[string, *netboxdiodeplugin.Branch](1, nil, DefaultBranchCacheTTL) + return &Ops{ - repository: repository, - nbClient: nbClient, - logger: logger, - limits: limits, + repository: repository, + nbClient: nbClient, + logger: logger, + limits: limits, + branchCache: branchCache, } } +// DefaultBranch fetches the default branch from the NetBox plugin with caching +func (o *Ops) DefaultBranch(ctx context.Context) (*netboxdiodeplugin.Branch, error) { + const cacheKey = "default_branch" + + // Check cache first + if cachedBranch, ok := o.branchCache.Get(cacheKey); ok { + o.logger.Debug("using cached default branch", "branch", cachedBranch) + return cachedBranch, nil + } + + // Cache miss - fetch from NetBox plugin + o.logger.Debug("cache miss - fetching default branch from NetBox plugin") + branch, err := o.nbClient.GetDefaultBranch(ctx) + if err != nil { + // Cache nil result for 404s (endpoint doesn't exist on older plugin versions) + // This prevents hammering the NetBox plugin with requests that will always fail + if errors.Is(err, netboxdiodeplugin.ErrDefaultBranchNotFound) { + o.logger.Debug("default-branch endpoint not found (older plugin version), caching nil result", "ttl", "5m") + o.branchCache.Add(cacheKey, nil) + return nil, nil // Return nil branch without error (gracefully handle missing endpoint) + } + return nil, err + } + + // Store in cache (automatically expires after 5 minutes) + o.branchCache.Add(cacheKey, branch) + o.logger.Debug("fetched and cached default branch", "branch", branch) + + return branch, nil +} + +// RefreshDefaultBranch forces a refresh of the default branch cache from NetBox +func (o *Ops) RefreshDefaultBranch(ctx context.Context) (*netboxdiodeplugin.Branch, error) { + const cacheKey = "default_branch" + o.branchCache.Remove(cacheKey) + return o.DefaultBranch(ctx) +} + // CreateIngestionLog creates a record for a newly received ingestion log func (o *Ops) CreateIngestionLog(ctx context.Context, ingestionLog *reconcilerpb.IngestionLog, sourceMetadata []byte) (*ops.CreateIngestionLogResult, error) { // TODO: this should be in a transaction. @@ -63,7 +116,20 @@ func (o *Ops) CreateIngestionLog(ctx context.Context, ingestionLog *reconcilerpb return nil, fmt.Errorf("failed to generate entity hash: %w", err) } - existingID, existingLog, err := o.repository.FindPriorIngestionLogByEntityHash(ctx, entityHash, nil) + // Fetch default branch from NetBox plugin (cached for 5 minutes) to ensure we search for prior ingestion logs in the correct branch context + var defaultBranchID *string + var branchIDForResult string + if branch, err := o.DefaultBranch(ctx); err != nil { + o.logger.Warn("failed to fetch default branch from NetBox plugin", "error", err) + // Continue with nil branch (main branch) if we can't fetch default branch + } else if branch != nil { + branchID := fmt.Sprintf("%s (%s)", branch.Name, branch.ID) + defaultBranchID = &branchID + branchIDForResult = branch.ID // Store the schema_id for GenerateChangeSet + o.logger.Debug("using default branch for ingestion log deduplication", "branch", branch.Name, "branchID", branch.ID) + } + + existingID, existingLog, err := o.repository.FindPriorIngestionLogByEntityHash(ctx, entityHash, defaultBranchID) if err != nil && !errors.Is(err, sql.ErrNoRows) { return nil, fmt.Errorf("failed to search for prior deviation: %w", err) } @@ -77,6 +143,7 @@ func (o *Ops) CreateIngestionLog(ctx context.Context, ingestionLog *reconcilerpb result := &ops.CreateIngestionLogResult{ ID: *id, IngestionLog: ingestionLog, + BranchID: branchIDForResult, } return result, nil } @@ -90,6 +157,7 @@ func (o *Ops) CreateIngestionLog(ctx context.Context, ingestionLog *reconcilerpb ID: *existingID, IngestionLog: existingLog, WasDuplicate: true, + BranchID: branchIDForResult, } return result, nil diff --git a/diode-server/reconciler/ops/types.go b/diode-server/reconciler/ops/types.go index a7e2252f..8987eec5 100644 --- a/diode-server/reconciler/ops/types.go +++ b/diode-server/reconciler/ops/types.go @@ -6,5 +6,6 @@ import "github.com/netboxlabs/diode/diode-server/gen/diode/v1/reconcilerpb" type CreateIngestionLogResult struct { ID int32 IngestionLog *reconcilerpb.IngestionLog - WasDuplicate bool // true if the ingestion log was a duplicate, in this case the prior ingestion log is returned + WasDuplicate bool // true if the ingestion log was a duplicate, in this case the prior ingestion log is returned + BranchID string // the branch ID used for this ingestion log (empty string means main branch) } diff --git a/diode-server/reconciler/ops_test.go b/diode-server/reconciler/ops_test.go index 2d9483f4..6b210896 100644 --- a/diode-server/reconciler/ops_test.go +++ b/diode-server/reconciler/ops_test.go @@ -304,6 +304,9 @@ func TestOpsCreateIngestionLog(t *testing.T) { mockNetBoxClient := pluginmocks.NewNetBoxAPI(t) opsInstance := reconciler.NewOps(mockRepository, mockNetBoxClient, logger, nil) + // Mock GetDefaultBranch to return nil (no default branch) + mockNetBoxClient.EXPECT().GetDefaultBranch(mock.Anything).Return((*netboxdiodeplugin.Branch)(nil), nil) + mockRepository.EXPECT().FindPriorIngestionLogByEntityHash(mock.Anything, mock.AnythingOfType("string"), (*string)(nil)). Return(tt.mockFindPriorIngestionLogID, tt.mockFindPriorIngestionLog, tt.mockFindPriorIngestionLogError) @@ -336,3 +339,225 @@ func TestOpsCreateIngestionLog(t *testing.T) { }) } } + +func TestOpsRefreshDefaultBranch(t *testing.T) { + logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug, AddSource: false})) + ctx := context.Background() + + tests := []struct { + name string + initialBranch *netboxdiodeplugin.Branch + initialError error + refreshedBranch *netboxdiodeplugin.Branch + refreshedError error + expectInitialCall bool + expectRefreshCall bool + expectCachedAfter bool + }{ + { + name: "refresh invalidates cache and fetches new data", + initialBranch: &netboxdiodeplugin.Branch{ + ID: "initial-branch", + Name: "Initial Branch", + }, + initialError: nil, + refreshedBranch: &netboxdiodeplugin.Branch{ + ID: "refreshed-branch", + Name: "Refreshed Branch", + }, + refreshedError: nil, + expectInitialCall: true, + expectRefreshCall: true, + expectCachedAfter: true, + }, + { + name: "refresh clears 404 cached nil and retries", + initialBranch: nil, + initialError: netboxdiodeplugin.ErrDefaultBranchNotFound, + refreshedBranch: &netboxdiodeplugin.Branch{ + ID: "new-branch", + Name: "New Branch", + }, + refreshedError: nil, + expectInitialCall: true, + expectRefreshCall: true, + expectCachedAfter: true, + }, + { + name: "refresh handles errors gracefully", + initialBranch: &netboxdiodeplugin.Branch{ + ID: "initial-branch", + Name: "Initial Branch", + }, + initialError: nil, + refreshedBranch: nil, + refreshedError: fmt.Errorf("temporary network error"), + expectInitialCall: true, + expectRefreshCall: true, + expectCachedAfter: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mockRepository := mocks.NewRepository(t) + mockNetBoxClient := pluginmocks.NewNetBoxAPI(t) + opsInstance := reconciler.NewOps(mockRepository, mockNetBoxClient, logger, nil) + + // Initial call - populate cache + if tt.expectInitialCall { + mockNetBoxClient.EXPECT().GetDefaultBranch(ctx).Return(tt.initialBranch, tt.initialError).Once() + } + + // Call DefaultBranch to populate cache + initialBranch, initialErr := opsInstance.DefaultBranch(ctx) + if tt.initialError == netboxdiodeplugin.ErrDefaultBranchNotFound { + require.NoError(t, initialErr) + require.Nil(t, initialBranch) + } else if tt.initialError != nil { + require.Error(t, initialErr) + } else { + require.NoError(t, initialErr) + require.Equal(t, tt.initialBranch, initialBranch) + } + + // Refresh call - should invalidate cache and fetch new data + if tt.expectRefreshCall { + mockNetBoxClient.EXPECT().GetDefaultBranch(ctx).Return(tt.refreshedBranch, tt.refreshedError).Once() + } + + refreshedBranch, refreshErr := opsInstance.RefreshDefaultBranch(ctx) + if tt.refreshedError != nil && tt.refreshedError != netboxdiodeplugin.ErrDefaultBranchNotFound { + require.Error(t, refreshErr) + } else { + require.NoError(t, refreshErr) + if tt.refreshedBranch != nil { + require.Equal(t, tt.refreshedBranch, refreshedBranch) + } + } + + // Verify cache after refresh - should use cached value on next call + if tt.expectCachedAfter && tt.refreshedError == nil { + // This call should NOT hit the API (using cache) + cachedBranch, cachedErr := opsInstance.DefaultBranch(ctx) + require.NoError(t, cachedErr) + require.Equal(t, refreshedBranch, cachedBranch) + } + + mockNetBoxClient.AssertExpectations(t) + }) + } +} + +func TestOpsDefaultBranch404Caching(t *testing.T) { + logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug, AddSource: false})) + ctx := context.Background() + + tests := []struct { + name string + firstCallError error + secondCallError error + expectedFirstCall bool + expectedSecondCall bool + expectBranchNil bool + expectErrorOnSecond bool + }{ + { + name: "404 error is cached - prevents repeated calls", + firstCallError: netboxdiodeplugin.ErrDefaultBranchNotFound, + secondCallError: nil, // Should not be called + expectedFirstCall: true, + expectedSecondCall: false, // Should use cache + expectBranchNil: true, + expectErrorOnSecond: false, + }, + { + name: "non-404 error is not cached - retries on second call", + firstCallError: fmt.Errorf("get default branch failed with status 500: Internal Server Error"), + secondCallError: fmt.Errorf("get default branch failed with status 500: Internal Server Error"), + expectedFirstCall: true, + expectedSecondCall: true, // Should retry + expectBranchNil: true, + expectErrorOnSecond: false, // Error is logged as warning, operation continues + }, + { + name: "successful response is cached", + firstCallError: nil, + secondCallError: nil, + expectedFirstCall: true, + expectedSecondCall: false, // Should use cache + expectBranchNil: false, + expectErrorOnSecond: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mockRepository := mocks.NewRepository(t) + mockNetBoxClient := pluginmocks.NewNetBoxAPI(t) + opsInstance := reconciler.NewOps(mockRepository, mockNetBoxClient, logger, nil) + + var firstBranch *netboxdiodeplugin.Branch + if tt.firstCallError == nil { + firstBranch = &netboxdiodeplugin.Branch{ + ID: "test-branch-id", + Name: "test-branch", + } + } + + // Mock repository methods for CreateIngestionLog + mockRepository.EXPECT().FindPriorIngestionLogByEntityHash(mock.Anything, mock.AnythingOfType("string"), mock.Anything). + Return(nil, nil, sql.ErrNoRows).Maybe() + mockRepository.EXPECT().CreateIngestionLog(mock.Anything, mock.Anything, mock.Anything, mock.AnythingOfType("string")). + Return(func() *int32 { id := int32(1); return &id }(), nil).Maybe() + + // Mock first call + firstCall := mockNetBoxClient.EXPECT().GetDefaultBranch(mock.Anything).Return(firstBranch, tt.firstCallError).Once() + + // Mock second call only if expected + if tt.expectedSecondCall { + mockNetBoxClient.EXPECT().GetDefaultBranch(mock.Anything).Return(firstBranch, tt.secondCallError).Once().NotBefore(firstCall) + } + + // First call - should always hit the API + branch1, err1 := opsInstance.CreateIngestionLog(ctx, &pb.IngestionLog{ + Id: "test-log-1", + ObjectType: "dcim.site", + Entity: &diodepb.Entity{ + Entity: &diodepb.Entity_Site{ + Site: &diodepb.Site{Name: "test-site-1"}, + }, + }, + }, nil) + + // Second call - behavior depends on caching + branch2, err2 := opsInstance.CreateIngestionLog(ctx, &pb.IngestionLog{ + Id: "test-log-2", + ObjectType: "dcim.site", + Entity: &diodepb.Entity{ + Entity: &diodepb.Entity_Site{ + Site: &diodepb.Site{Name: "test-site-2"}, + }, + }, + }, nil) + + // Verify results - all calls should succeed (errors are gracefully handled) + require.NoError(t, err1) + require.NoError(t, err2) + require.NotNil(t, branch1) + require.NotNil(t, branch2) + + // Verify branch caching worked correctly + if tt.expectBranchNil { + require.Empty(t, branch1.BranchID) + require.Empty(t, branch2.BranchID) + } else { + require.Equal(t, "test-branch-id", branch1.BranchID) + require.Equal(t, "test-branch-id", branch2.BranchID) + } + + // Mock assertions verify the expected number of calls were made + mockNetBoxClient.AssertExpectations(t) + }) + } +} diff --git a/diode-server/telemetry/mocks/metricrecorder.go b/diode-server/telemetry/mocks/metricrecorder.go index 1d9001d8..0b6f892a 100644 --- a/diode-server/telemetry/mocks/metricrecorder.go +++ b/diode-server/telemetry/mocks/metricrecorder.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.53.5. DO NOT EDIT. package mocks