mirror of
https://github.com/SigNoz/signoz.git
synced 2026-09-28 14:20:42 +01:00
Compare commits
1 Commits
fix/opaque
...
v0.143.0-c
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ec05bfe755 |
@@ -364,26 +364,12 @@ func (provider *provider) gc(ctx context.Context, org *types.Organization) error
|
||||
}
|
||||
|
||||
func (provider *provider) flushLastObservedAt(ctx context.Context, org *types.Organization) error {
|
||||
tokens, err := provider.tokenStore.ListByOrgID(ctx, org.ID)
|
||||
accessTokenToLastObservedAt, err := provider.listLastObservedAtDesc(ctx, org.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
observedTokens := make([]*authtypes.StorableToken, 0, len(tokens))
|
||||
for _, token := range tokens {
|
||||
cachedLastObservedAt, ok := provider.lastObservedAtCache.Get(lastObservedAtCacheKey(token.AccessToken, token.UserID))
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
if err := token.UpdateLastObservedAt(cachedLastObservedAt); err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
observedTokens = append(observedTokens, token)
|
||||
}
|
||||
|
||||
if err := provider.tokenStore.UpdateLastObservedAt(ctx, observedTokens); err != nil {
|
||||
if err := provider.tokenStore.UpdateLastObservedAtByAccessToken(ctx, accessTokenToLastObservedAt); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
|
||||
@@ -232,16 +232,15 @@ func (store *store) ListByUserID(ctx context.Context, userID valuer.UUID) ([]*au
|
||||
return tokens, nil
|
||||
}
|
||||
|
||||
func (store *store) UpdateLastObservedAt(ctx context.Context, tokens []*authtypes.StorableToken) error {
|
||||
if len(tokens) == 0 {
|
||||
func (store *store) UpdateLastObservedAtByAccessToken(ctx context.Context, accessTokenToLastObservedAt []map[string]any) error {
|
||||
if len(accessTokenToLastObservedAt) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
values := store.
|
||||
sqlstore.
|
||||
BunDBCtx(ctx).
|
||||
NewValues(&tokens).
|
||||
Column("id", "last_observed_at", "updated_at")
|
||||
NewValues(&accessTokenToLastObservedAt)
|
||||
|
||||
_, err := store.
|
||||
sqlstore.
|
||||
@@ -251,8 +250,8 @@ func (store *store) UpdateLastObservedAt(ctx context.Context, tokens []*authtype
|
||||
Model((*authtypes.StorableToken)(nil)).
|
||||
TableExpr("update_cte").
|
||||
Set("last_observed_at = update_cte.last_observed_at").
|
||||
Set("updated_at = update_cte.updated_at").
|
||||
Where("auth_token.id = update_cte.id").
|
||||
Where("auth_token.access_token = update_cte.access_token").
|
||||
Where("auth_token.user_id = update_cte.user_id").
|
||||
Exec(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
@@ -1,74 +0,0 @@
|
||||
package sqltokenizerstore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/DATA-DOG/go-sqlmock"
|
||||
"github.com/SigNoz/signoz/pkg/sqlstore"
|
||||
"github.com/SigNoz/signoz/pkg/sqlstore/sqlstoretest"
|
||||
"github.com/SigNoz/signoz/pkg/types/authtypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestUpdateLastObservedAt(t *testing.T) {
|
||||
testCases := []struct {
|
||||
name string
|
||||
provider string
|
||||
tokens []*authtypes.StorableToken
|
||||
expectedQuery string
|
||||
}{
|
||||
{
|
||||
name: "Sqlite_Empty",
|
||||
provider: "sqlite",
|
||||
tokens: nil,
|
||||
expectedQuery: "",
|
||||
},
|
||||
{
|
||||
name: "Postgres_Empty",
|
||||
provider: "postgres",
|
||||
tokens: []*authtypes.StorableToken{},
|
||||
expectedQuery: "",
|
||||
},
|
||||
{
|
||||
name: "Sqlite_OneToken",
|
||||
provider: "sqlite",
|
||||
tokens: []*authtypes.StorableToken{
|
||||
{ID: valuer.MustNewUUID("019984d1-0000-7000-8000-000000000001"), AccessToken: "access-one", RefreshToken: "refresh-one", LastObservedAt: time.Date(2026, 9, 22, 10, 0, 0, 0, time.UTC), UpdatedAt: time.Date(2026, 9, 22, 10, 0, 1, 0, time.UTC)},
|
||||
},
|
||||
expectedQuery: `WITH "update_cte" ("id", "last_observed_at", "updated_at") AS (VALUES ('019984d1-0000-7000-8000-000000000001', '2026-09-22 10:00:00+00:00', '2026-09-22 10:00:01+00:00')) UPDATE "auth_token" AS "auth_token" SET last_observed_at = update_cte.last_observed_at, updated_at = update_cte.updated_at FROM update_cte WHERE (auth_token.id = update_cte.id)`,
|
||||
},
|
||||
{
|
||||
name: "Postgres_TwoTokens",
|
||||
provider: "postgres",
|
||||
tokens: []*authtypes.StorableToken{
|
||||
{ID: valuer.MustNewUUID("019984d1-0000-7000-8000-000000000002"), AccessToken: "access-two", RefreshToken: "refresh-two", LastObservedAt: time.Date(2026, 9, 22, 11, 0, 0, 0, time.UTC), UpdatedAt: time.Date(2026, 9, 22, 11, 0, 1, 0, time.UTC)},
|
||||
{ID: valuer.MustNewUUID("019984d1-0000-7000-8000-000000000003"), AccessToken: "access-three", RefreshToken: "refresh-three", LastObservedAt: time.Date(2026, 9, 22, 12, 0, 0, 0, time.UTC), UpdatedAt: time.Date(2026, 9, 22, 12, 0, 1, 0, time.UTC)},
|
||||
},
|
||||
expectedQuery: `WITH "update_cte" ("id", "last_observed_at", "updated_at") AS (VALUES ('019984d1-0000-7000-8000-000000000002'::text, '2026-09-22 11:00:00+00:00'::TIMESTAMPTZ, '2026-09-22 11:00:01+00:00'::TIMESTAMPTZ), ('019984d1-0000-7000-8000-000000000003'::text, '2026-09-22 12:00:00+00:00'::TIMESTAMPTZ, '2026-09-22 12:00:01+00:00'::TIMESTAMPTZ)) UPDATE "auth_token" AS "auth_token" SET last_observed_at = update_cte.last_observed_at, updated_at = update_cte.updated_at FROM update_cte WHERE (auth_token.id = update_cte.id)`,
|
||||
},
|
||||
}
|
||||
|
||||
for _, testCase := range testCases {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
var executedQuery string
|
||||
matcher := sqlmock.QueryMatcherFunc(func(_, actual string) error {
|
||||
executedQuery = actual
|
||||
return nil
|
||||
})
|
||||
|
||||
sqlStore := sqlstoretest.New(sqlstore.Config{Provider: testCase.provider}, matcher)
|
||||
if testCase.expectedQuery != "" {
|
||||
sqlStore.Mock().ExpectExec("").WillReturnResult(sqlmock.NewResult(0, int64(len(testCase.tokens))))
|
||||
}
|
||||
|
||||
err := NewStore(sqlStore).UpdateLastObservedAt(context.Background(), testCase.tokens)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, sqlStore.Mock().ExpectationsWereMet())
|
||||
assert.Equal(t, testCase.expectedQuery, executedQuery)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -258,6 +258,6 @@ type TokenStore interface {
|
||||
// Delete a token by userID.
|
||||
DeleteByUserID(context.Context, valuer.UUID) error
|
||||
|
||||
// Update last observed at of the given tokens.
|
||||
UpdateLastObservedAt(context.Context, []*StorableToken) error
|
||||
// Update last observed at by access token.
|
||||
UpdateLastObservedAtByAccessToken(context.Context, []map[string]any) error
|
||||
}
|
||||
|
||||
@@ -208,6 +208,35 @@ func NewGettableUnmappedModels(items []*UnmappedModel) *GettableUnmappedModels {
|
||||
}
|
||||
}
|
||||
|
||||
func (u *UpdatableLLMPricingRule) UnmarshalJSON(data []byte) error {
|
||||
type Alias UpdatableLLMPricingRule
|
||||
|
||||
var temp Alias
|
||||
if err := json.Unmarshal(data, &temp); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
*u = UpdatableLLMPricingRule(temp)
|
||||
return u.Validate()
|
||||
}
|
||||
|
||||
// Validate mirrors the collector's pattern check: at least one pattern, none
|
||||
// empty, all valid path.Match globs.
|
||||
func (u *UpdatableLLMPricingRule) Validate() error {
|
||||
if len(u.ModelPattern) == 0 {
|
||||
return errors.Newf(errors.TypeInvalidInput, ErrCodePricingRuleInvalidInput, "model %q: modelPattern must contain at least one pattern", u.Model)
|
||||
}
|
||||
for _, p := range u.ModelPattern {
|
||||
if p == "" {
|
||||
return errors.Newf(errors.TypeInvalidInput, ErrCodePricingRuleInvalidInput, "model %q: modelPattern must not contain an empty pattern", u.Model)
|
||||
}
|
||||
if _, err := path.Match(p, ""); err != nil {
|
||||
return errors.Newf(errors.TypeInvalidInput, ErrCodePricingRuleInvalidInput, "model %q: modelPattern %q is not a valid glob", u.Model, p)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func NewLLMPricingRuleFromUpdatable(u *UpdatableLLMPricingRule, orgID valuer.UUID, userEmail string, now time.Time) *LLMPricingRule {
|
||||
id := valuer.GenerateUUID()
|
||||
if u.ID != nil {
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package llmpricingruletypes
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
@@ -126,3 +127,34 @@ func TestGenerateCollectorConfig_EmptyInputPassthrough(t *testing.T) {
|
||||
assert.Equal(t, in, out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpdatableLLMPricingRuleUnmarshalJSON(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
pattern string
|
||||
wantErr bool
|
||||
}{
|
||||
{name: "valid", pattern: `["gpt-4o*", "gpt-4o"]`},
|
||||
{name: "missing", pattern: ``, wantErr: true},
|
||||
{name: "null", pattern: `null`, wantErr: true},
|
||||
{name: "empty_list", pattern: `[]`, wantErr: true},
|
||||
{name: "empty_entry", pattern: `["gpt-4o*", ""]`, wantErr: true},
|
||||
{name: "bad_glob", pattern: `["gpt-["]`, wantErr: true},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
body := `{"modelName": "gpt-4o"}`
|
||||
if tc.pattern != "" {
|
||||
body = `{"modelName": "gpt-4o", "modelPattern": ` + tc.pattern + `}`
|
||||
}
|
||||
var req UpdatableLLMPricingRules
|
||||
err := json.Unmarshal([]byte(`{"rules": [`+body+`]}`), &req)
|
||||
if tc.wantErr {
|
||||
assert.Error(t, err)
|
||||
} else {
|
||||
assert.NoError(t, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -130,3 +130,17 @@ def test_bulk_sync(
|
||||
assert all(r["pricing"]["input"] == 5 for r in stored)
|
||||
|
||||
delete_all_llm_pricing_rules(signoz, token)
|
||||
|
||||
|
||||
def test_rejects_rule_without_pattern(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: types.Operation, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
):
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
delete_all_llm_pricing_rules(signoz, token)
|
||||
|
||||
rules = zeus_rules(10)
|
||||
rules[1]["modelPattern"] = []
|
||||
assert upsert_llm_pricing_rules(signoz, token, rules).status_code == HTTPStatus.BAD_REQUEST
|
||||
assert list_llm_pricing_rules(signoz, token) == []
|
||||
|
||||
@@ -1,36 +0,0 @@
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from http import HTTPStatus
|
||||
|
||||
import requests
|
||||
from sqlalchemy import sql
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
|
||||
|
||||
def test_last_observed_at_is_flushed(signoz: types.SigNoz, get_token: Callable[[str, str], str]) -> None:
|
||||
"""Verify the tokenizer GC persists the cached last observed at of a used token to the sql store."""
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
response = requests.get(
|
||||
signoz.self.host_configs["8080"].get("/api/v2/users/me"),
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
timeout=5,
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
|
||||
deadline = time.time() + 30
|
||||
while time.time() < deadline:
|
||||
with signoz.sqlstore.conn.connect() as conn:
|
||||
row = conn.execute(
|
||||
sql.text("SELECT last_observed_at FROM auth_token WHERE access_token = :access_token"),
|
||||
{"access_token": token},
|
||||
).fetchone()
|
||||
|
||||
if row is not None and row[0] is not None:
|
||||
return
|
||||
|
||||
time.sleep(1)
|
||||
|
||||
raise AssertionError("last_observed_at was not flushed to the sql store within 30s")
|
||||
@@ -1,33 +0,0 @@
|
||||
import pytest
|
||||
from testcontainers.core.container import Network
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.signoz import create_signoz
|
||||
|
||||
|
||||
@pytest.fixture(name="signoz", scope="package")
|
||||
def signoz_passwordauthn(
|
||||
network: Network,
|
||||
zeus: types.TestContainerDocker,
|
||||
gateway: types.TestContainerDocker,
|
||||
sqlstore: types.TestContainerSQL,
|
||||
clickhouse: types.TestContainerClickhouse,
|
||||
request: pytest.FixtureRequest,
|
||||
pytestconfig: pytest.Config,
|
||||
) -> types.SigNoz:
|
||||
"""
|
||||
Package-scoped fixture for SigNoz with a short tokenizer GC interval so the last observed at flush runs within a test.
|
||||
"""
|
||||
return create_signoz(
|
||||
network=network,
|
||||
zeus=zeus,
|
||||
gateway=gateway,
|
||||
sqlstore=sqlstore,
|
||||
clickhouse=clickhouse,
|
||||
request=request,
|
||||
pytestconfig=pytestconfig,
|
||||
cache_key="signoz-passwordauthn",
|
||||
env_overrides={
|
||||
"SIGNOZ_TOKENIZER_OPAQUE_GC_INTERVAL": "5s",
|
||||
},
|
||||
)
|
||||
Reference in New Issue
Block a user