mirror of
https://github.com/SigNoz/signoz.git
synced 2026-09-07 20:10:42 +01:00
Compare commits
1 Commits
ns/harness
...
chore/remo
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b2a6e1b786 |
@@ -1,7 +0,0 @@
|
||||
{
|
||||
"$schema": "https://opencode.ai/config.json",
|
||||
"lsp": true,
|
||||
"experimental": {
|
||||
"disable_paste_summary": true
|
||||
}
|
||||
}
|
||||
@@ -7,6 +7,5 @@ Applies to everything in the repo — code, config, workflows.
|
||||
- **Rationale goes in prose, not source.** Why a version is pinned, why a job exists, how a subsystem fits together — that belongs in the README or the PR.
|
||||
- **Never remove pre-existing comments** when editing code. The bar above applies to comments you write, not comments already there.
|
||||
- **Never talk to the reviewer.** No comments about where a change came from, what was changed, or why the change is correct — that belongs in the PR description and is noise the moment it merges.
|
||||
- **Less is more.** When writing something intended for human consumption, (comment, commit message, reply to prompt) use as few words as possible. Pick every word meticulously to reduce the volume to a strict minimum. Be down to the point. Less is more.
|
||||
|
||||
Language rules build on this one: [`go-comments`](go-comments.md), [`py-comments`](py-comments.md).
|
||||
|
||||
@@ -1,16 +0,0 @@
|
||||
---
|
||||
paths:
|
||||
- "**/*.go"
|
||||
---
|
||||
|
||||
# Contribution guidelines
|
||||
|
||||
- When making Go changes, always ensure they follow the contributing guildelines in [`docs/contributing/go/`](../../docs/contributing/go/).
|
||||
- Look for existing patterns in the codebase for any change before implementing the changes.
|
||||
- If any API contract is modified, generate the OpenAPI specs with `make gen-openapi-specs`.
|
||||
- Always keep the OpenAPI spec generated in a separate commit, so the whole commit can be dropped in case of conflicts during merge. Do not try to resolve conflict in generated files, instead just generate them again.
|
||||
- Avoid breaking function calls unncessarily into multilines for couple of arguments.
|
||||
- Try to keep most computational only logic in types package itself related to a domain type, use modules as the orchestraction layer cordinating different layers and all db queries in store layer. Check the serviceaccount modules for inspiration when confused.
|
||||
- When defining types, keep the structure of file to have any constants and variables first, then exported types and exported methods and then finally the unexported types and methods.
|
||||
- Never import types or other modules in migration files, duplicate the required type or method to keep migration free from changes.
|
||||
- Always run the gofmt tool for formating beforing commiting any changes.
|
||||
@@ -2,10 +2,6 @@
|
||||
|
||||
- **Follow the template** (`.github/pull_request_template.md`): fill in its headings (Description / Issues closed by this PR / Screenshots / Additional Information). Don't add sections the template doesn't have.
|
||||
- **Keep only the headings that apply.** Delete every heading that has nothing under it, along with its `<!--...-->` placeholder comment. The body must never contain an empty heading — if only Description applies, the body has exactly that one heading.
|
||||
- **Keep the description concise and human-readable.** A few non repetitive bullets saying what changed and why, for a reviewer skimming it — not a wall of text, not a restatement of the diff, not generated boilerplate and not the user agent conversation details.
|
||||
- **Keep the description concise and human-readable.** A few plain bullets saying what changed and why, for a reviewer skimming it — not a wall of text, not a restatement of the diff, not generated boilerplate.
|
||||
- **Reference issues with `Closes #issue-number`** under "Issues closed by this PR" so they auto-close on merge. This goes in the PR description only — never in commit messages.
|
||||
- **Breaking changes can be added in additional information section** if any.
|
||||
- **AI assistance in commits may optionally be disclosed with an `Assisted-by:` trailer** naming the model (e.g. `Assisted-by: Claude Opus 4.5`) — do NOT use a `Co-authored-by:` trailer for this.
|
||||
- **Keep the commit body short and human readable** focused on decision made if any. Commit body must not re-iterate the changes done, skip if title is sufficient in conveying the change.
|
||||
- **Use convensional commit format** for commits and PR title.
|
||||
- **Do not amend the commits once pushed.** Always create a new commit once changes are pushed to remote.
|
||||
|
||||
1
.gitignore
vendored
1
.gitignore
vendored
@@ -232,4 +232,3 @@ pyrightconfig.json
|
||||
# dev
|
||||
.dev/
|
||||
.claude/worktrees/
|
||||
.claude/settings.local.json
|
||||
|
||||
34
Makefile
34
Makefile
@@ -81,13 +81,10 @@ devenv-clickhouse-clean: ## Clean all ClickHouse data from filesystem
|
||||
##############################################################
|
||||
# go commands
|
||||
##############################################################
|
||||
SIGNOZ_SQLSTORE_SQLITE_PATH ?= signoz.db
|
||||
SIGNOZ_APISERVER_ADDRESS ?= 0.0.0.0:8080
|
||||
|
||||
.PHONY: go-run-enterprise
|
||||
go-run-enterprise: ## Runs the enterprise go backend server
|
||||
@SIGNOZ_INSTRUMENTATION_LOGS_LEVEL=debug \
|
||||
SIGNOZ_SQLSTORE_SQLITE_PATH=$(SIGNOZ_SQLSTORE_SQLITE_PATH) \
|
||||
SIGNOZ_SQLSTORE_SQLITE_PATH=signoz.db \
|
||||
SIGNOZ_WEB_ENABLED=false \
|
||||
SIGNOZ_TOKENIZER_JWT_SECRET=secret \
|
||||
SIGNOZ_ALERTMANAGER_PROVIDER=signoz \
|
||||
@@ -104,7 +101,7 @@ go-test: ## Runs go unit tests
|
||||
.PHONY: go-run-community
|
||||
go-run-community: ## Runs the community go backend server
|
||||
@SIGNOZ_INSTRUMENTATION_LOGS_LEVEL=debug \
|
||||
SIGNOZ_SQLSTORE_SQLITE_PATH=$(SIGNOZ_SQLSTORE_SQLITE_PATH) \
|
||||
SIGNOZ_SQLSTORE_SQLITE_PATH=signoz.db \
|
||||
SIGNOZ_WEB_ENABLED=false \
|
||||
SIGNOZ_TOKENIZER_JWT_SECRET=secret \
|
||||
SIGNOZ_ALERTMANAGER_PROVIDER=signoz \
|
||||
@@ -114,28 +111,6 @@ go-run-community: ## Runs the community go backend server
|
||||
go run -race \
|
||||
$(GO_BUILD_CONTEXT_COMMUNITY)/*.go server
|
||||
|
||||
.PHONY: go-stop
|
||||
go-stop: ## Stops the go backend server listening on SIGNOZ_APISERVER_ADDRESS, waiting for it to release every port it holds
|
||||
@PORT=$(lastword $(subst :, ,$(SIGNOZ_APISERVER_ADDRESS))); \
|
||||
PIDS=$$(lsof -ti tcp:$$PORT); \
|
||||
if [ -z "$$PIDS" ]; then \
|
||||
echo "No signoz server running on port $$PORT."; \
|
||||
echo "If it's running on a different port, rerun as: make go-stop SIGNOZ_APISERVER_ADDRESS=host:port"; \
|
||||
exit 0; \
|
||||
fi; \
|
||||
kill $$PIDS 2>/dev/null; \
|
||||
for i in $$(seq 1 10); do \
|
||||
alive=$$(for p in $$PIDS; do kill -0 $$p 2>/dev/null && echo $$p; done); \
|
||||
[ -z "$$alive" ] && break; \
|
||||
sleep 1; \
|
||||
done; \
|
||||
alive=$$(for p in $$PIDS; do kill -0 $$p 2>/dev/null && echo $$p; done); \
|
||||
if [ -n "$$alive" ]; then \
|
||||
echo "Graceful shutdown did not finish in 10s, sending SIGKILL to $$alive"; \
|
||||
kill -9 $$alive 2>/dev/null; \
|
||||
fi; \
|
||||
echo "Stopped signoz server on port $$PORT (pid $$PIDS)"
|
||||
|
||||
.PHONY: go-build-community $(GO_BUILD_ARCHS_COMMUNITY)
|
||||
go-build-community: ## Builds the go backend server for community
|
||||
go-build-community: $(GO_BUILD_ARCHS_COMMUNITY)
|
||||
@@ -266,8 +241,3 @@ semconv-generate: ## Regenerate semantic-convention families for Go and TypeScri
|
||||
gen-mocks:
|
||||
@echo ">> Generating mocks"
|
||||
@mockery --config .mockery.yml
|
||||
|
||||
.PHONY: gen-openapi-specs
|
||||
gen-openapi-specs:
|
||||
@go run cmd/enterprise/*.go generate openapi
|
||||
cd frontend && pnpm generate:api && cd -
|
||||
|
||||
@@ -138,8 +138,6 @@ sqlstore:
|
||||
|
||||
##################### APIServer #####################
|
||||
apiserver:
|
||||
# The TCP address the API server listens on, in the form "host:port".
|
||||
address: 0.0.0.0:8080
|
||||
timeout:
|
||||
# Default request timeout.
|
||||
default: 60s
|
||||
@@ -399,7 +397,6 @@ identn:
|
||||
# headers to use for tokenizer identN resolver
|
||||
headers:
|
||||
- Authorization
|
||||
- Sec-WebSocket-Protocol
|
||||
apikey:
|
||||
# toggle apikey identN
|
||||
enabled: true
|
||||
|
||||
@@ -83,13 +83,7 @@ This command:
|
||||
|
||||
You should see: `{"status":"ok"}`
|
||||
|
||||
3. Stop it when you're done:
|
||||
```bash
|
||||
make go-stop
|
||||
```
|
||||
|
||||
> 💡 **Tip**: The API server runs at `http://localhost:8080/` by default. You can configure this using `apiserver.address` configuration option. See
|
||||
> [running more than one instance](#how-do-i-run-more-than-one-instance) if you need that for agentic testing.
|
||||
> 💡 **Tip**: The API server runs at `http://localhost:8080/` by default
|
||||
|
||||
### 4. Setting up the Frontend
|
||||
|
||||
@@ -125,36 +119,6 @@ To verify everything is working correctly:
|
||||
3. **Check Backend**: `curl http://localhost:8080/api/v1/health` (should return `{"status":"ok"}`)
|
||||
4. **Check Frontend**: Open `http://localhost:3301` in your browser
|
||||
|
||||
## How do I run more than one instance?
|
||||
|
||||
Handy when you keep several branches checked out as separate git worktrees. Every port
|
||||
and path below is read from the environment, so set them on the `make` call:
|
||||
|
||||
```bash
|
||||
SIGNOZ_APISERVER_ADDRESS=0.0.0.0:8081 \
|
||||
SIGNOZ_SQLSTORE_SQLITE_PATH=/path/to/main/sqlite.db \
|
||||
SIGNOZ_INSTRUMENTATION_METRICS_READERS_PULL_EXPORTER_PROMETHEUS_PORT=9091 \
|
||||
make go-run-community
|
||||
```
|
||||
|
||||
| Variable | Default | Why you'd change it |
|
||||
| --- | --- | --- |
|
||||
| `SIGNOZ_APISERVER_ADDRESS` | `0.0.0.0:8080` | Address the API server listens on |
|
||||
| `SIGNOZ_SQLSTORE_SQLITE_PATH` | `signoz.db` in worktree | To reuse same database |
|
||||
| `SIGNOZ_INSTRUMENTATION_METRICS_READERS_PULL_EXPORTER_PROMETHEUS_PORT` | `9090` | Bound by the Prometheus metrics exporter on startup |
|
||||
|
||||
Point the frontend at whichever backend you want, in `frontend/.env`:
|
||||
|
||||
```env
|
||||
VITE_FRONTEND_API_ENDPOINT=http://localhost:8081
|
||||
```
|
||||
|
||||
Stop an instance using the address it was started on:
|
||||
|
||||
```bash
|
||||
make go-stop SIGNOZ_APISERVER_ADDRESS=0.0.0.0:8081
|
||||
```
|
||||
|
||||
## How to send test data?
|
||||
|
||||
You can now send telemetry data to your local SigNoz instance:
|
||||
|
||||
@@ -3,30 +3,56 @@ package app
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"slices"
|
||||
|
||||
"go.opentelemetry.io/contrib/instrumentation/github.com/gorilla/mux/otelmux"
|
||||
"go.opentelemetry.io/otel/propagation"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
|
||||
"github.com/gorilla/handlers"
|
||||
|
||||
"github.com/rs/cors"
|
||||
"github.com/soheilhy/cmux"
|
||||
|
||||
"github.com/SigNoz/signoz/ee/query-service/app/api"
|
||||
"github.com/SigNoz/signoz/ee/query-service/usage"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/http/middleware"
|
||||
"github.com/SigNoz/signoz/pkg/signoz"
|
||||
"github.com/SigNoz/signoz/pkg/web"
|
||||
|
||||
"log/slog"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/query-service/agentConf"
|
||||
baseapp "github.com/SigNoz/signoz/pkg/query-service/app"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/app/clickhouseReader"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/app/integrations"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/app/logparsingpipeline"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/app/opamp"
|
||||
opAmpModel "github.com/SigNoz/signoz/pkg/query-service/app/opamp/model"
|
||||
baseconst "github.com/SigNoz/signoz/pkg/query-service/constants"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/healthcheck"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/utils"
|
||||
)
|
||||
|
||||
// Server runs auxiliary servers (opamp) alongside the signoz apiserver
|
||||
// Server runs HTTP, Mux and a grpc server
|
||||
type Server struct {
|
||||
config signoz.Config
|
||||
signoz *signoz.SigNoz
|
||||
|
||||
// public http router
|
||||
httpConn net.Listener
|
||||
httpServer *http.Server
|
||||
httpHostPort string
|
||||
|
||||
opampServer *opamp.Server
|
||||
|
||||
// Usage manager
|
||||
usageManager *usage.Manager
|
||||
|
||||
unavailableChannel chan healthcheck.Status
|
||||
}
|
||||
|
||||
// NewServer creates and initializes Server
|
||||
@@ -101,26 +127,22 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Register the legacy query-service routes on the apiserver router. The
|
||||
// apiserver owns the HTTP server and applies the middleware chain at serve
|
||||
// time, so these routes get the same treatment as the apiserver routes.
|
||||
r := signoz.APIServer.Router()
|
||||
am := middleware.NewAuthZ(signoz.Instrumentation.Logger(), signoz.Modules.OrgGetter, signoz.Authz)
|
||||
|
||||
apiHandler.RegisterRoutes(r, am)
|
||||
apiHandler.RegisterLogsRoutes(r, am)
|
||||
apiHandler.RegisterIntegrationRoutes(r, am)
|
||||
apiHandler.RegisterQueryRangeV3Routes(r, am)
|
||||
apiHandler.RegisterQueryRangeV4Routes(r, am)
|
||||
apiHandler.RegisterWebSocketPaths(r, am)
|
||||
apiHandler.RegisterMessagingQueuesRoutes(r, am)
|
||||
apiHandler.RegisterThirdPartyApiRoutes(r, am)
|
||||
apiHandler.RegisterTraceFunnelsRoutes(r, am)
|
||||
|
||||
s := &Server{
|
||||
usageManager: usageManager,
|
||||
config: config,
|
||||
signoz: signoz,
|
||||
httpHostPort: baseconst.HTTPHostPort,
|
||||
unavailableChannel: make(chan healthcheck.Status),
|
||||
usageManager: usageManager,
|
||||
}
|
||||
|
||||
httpServer, err := s.createPublicServer(apiHandler, signoz.Web)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
s.httpServer = httpServer
|
||||
|
||||
s.opampServer = opamp.InitializeServer(
|
||||
&opAmpModel.AllAgents, agentConfMgr, signoz.Instrumentation,
|
||||
)
|
||||
@@ -128,14 +150,132 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// Start starts the opamp websocket server. The HTTP API server is started by
|
||||
// the signoz registry.
|
||||
// HealthCheckStatus returns health check status channel a client can subscribe to
|
||||
func (s Server) HealthCheckStatus() chan healthcheck.Status {
|
||||
return s.unavailableChannel
|
||||
}
|
||||
|
||||
func (s *Server) createPublicServer(apiHandler *api.APIHandler, web web.Web) (*http.Server, error) {
|
||||
r := baseapp.NewRouter()
|
||||
am := middleware.NewAuthZ(s.signoz.Instrumentation.Logger(), s.signoz.Modules.OrgGetter, s.signoz.Authz)
|
||||
|
||||
r.Use(middleware.NewRecovery(s.signoz.Instrumentation.Logger()).Wrap)
|
||||
r.Use(otelmux.Middleware(
|
||||
"apiserver",
|
||||
otelmux.WithMeterProvider(s.signoz.Instrumentation.MeterProvider()),
|
||||
otelmux.WithTracerProvider(s.signoz.Instrumentation.TracerProvider()),
|
||||
otelmux.WithPropagators(propagation.NewCompositeTextMapPropagator(propagation.Baggage{}, propagation.TraceContext{})),
|
||||
otelmux.WithFilter(func(r *http.Request) bool {
|
||||
return !slices.Contains([]string{"/api/v1/health"}, r.URL.Path)
|
||||
}),
|
||||
))
|
||||
r.Use(middleware.NewIdentN(s.signoz.IdentNResolver, s.signoz.Sharder, s.signoz.Instrumentation.Logger()).Wrap)
|
||||
r.Use(middleware.NewTimeout(s.signoz.Instrumentation.Logger(),
|
||||
s.config.APIServer.Timeout.ExcludedRoutes,
|
||||
s.config.APIServer.Timeout.Default,
|
||||
s.config.APIServer.Timeout.Max,
|
||||
).Wrap)
|
||||
r.Use(middleware.NewResource(s.signoz.Instrumentation.Logger()).Wrap)
|
||||
r.Use(middleware.NewAudit(s.signoz.Instrumentation.Logger(), s.config.APIServer.Logging.ExcludedRoutes, s.signoz.Auditor).Wrap)
|
||||
r.Use(middleware.NewComment().Wrap)
|
||||
|
||||
apiHandler.RegisterRoutes(r, am)
|
||||
apiHandler.RegisterLogsRoutes(r, am)
|
||||
apiHandler.RegisterIntegrationRoutes(r, am)
|
||||
apiHandler.RegisterQueryRangeV3Routes(r, am)
|
||||
apiHandler.RegisterQueryRangeV4Routes(r, am)
|
||||
apiHandler.RegisterMessagingQueuesRoutes(r, am)
|
||||
apiHandler.RegisterThirdPartyApiRoutes(r, am)
|
||||
apiHandler.RegisterTraceFunnelsRoutes(r, am)
|
||||
|
||||
err := s.signoz.APIServer.AddToRouter(r)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
c := cors.New(cors.Options{
|
||||
AllowedOrigins: []string{"*"},
|
||||
AllowedMethods: []string{"GET", "DELETE", "POST", "PUT", "PATCH", "OPTIONS"},
|
||||
AllowedHeaders: []string{"Accept", "Authorization", "Content-Type", "cache-control"},
|
||||
})
|
||||
|
||||
handler := c.Handler(r)
|
||||
|
||||
handler = handlers.CompressHandler(handler)
|
||||
|
||||
err = web.AddToRouter(r)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
routePrefix := s.config.Global.ExternalPath()
|
||||
if routePrefix != "" {
|
||||
prefixed := http.StripPrefix(routePrefix, handler)
|
||||
handler = http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
|
||||
switch req.URL.Path {
|
||||
case "/api/v1/health", "/api/v2/healthz", "/api/v2/readyz", "/api/v2/livez":
|
||||
r.ServeHTTP(w, req)
|
||||
return
|
||||
}
|
||||
|
||||
prefixed.ServeHTTP(w, req)
|
||||
})
|
||||
}
|
||||
|
||||
return &http.Server{
|
||||
Handler: handler,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// initListeners initialises listeners of the server
|
||||
func (s *Server) initListeners() error {
|
||||
// listen on public port
|
||||
var err error
|
||||
publicHostPort := s.httpHostPort
|
||||
if publicHostPort == "" {
|
||||
return fmt.Errorf("baseconst.HTTPHostPort is required")
|
||||
}
|
||||
|
||||
s.httpConn, err = net.Listen("tcp", publicHostPort)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
slog.Info(fmt.Sprintf("Query server started listening on %s...", s.httpHostPort))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Start listening on http and private http port concurrently
|
||||
func (s *Server) Start(ctx context.Context) error {
|
||||
err := s.initListeners()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var httpPort int
|
||||
if port, err := utils.GetPort(s.httpConn.Addr()); err == nil {
|
||||
httpPort = port
|
||||
}
|
||||
|
||||
go func() {
|
||||
slog.Info("Starting HTTP server", "port", httpPort, "addr", s.httpHostPort)
|
||||
|
||||
switch err := s.httpServer.Serve(s.httpConn); err {
|
||||
case nil, http.ErrServerClosed, cmux.ErrListenerClosed:
|
||||
// normal exit, nothing to do
|
||||
default:
|
||||
slog.Error("Could not start HTTP server", errors.Attr(err))
|
||||
}
|
||||
s.unavailableChannel <- healthcheck.Unavailable
|
||||
}()
|
||||
|
||||
go func() {
|
||||
slog.Info("Starting OpAmp Websocket server", "addr", baseconst.OpAmpWsEndpoint)
|
||||
err := s.opampServer.Start(baseconst.OpAmpWsEndpoint)
|
||||
if err != nil {
|
||||
slog.Error("opamp ws server failed to start", errors.Attr(err))
|
||||
s.unavailableChannel <- healthcheck.Unavailable
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -143,6 +283,12 @@ func (s *Server) Start(ctx context.Context) error {
|
||||
}
|
||||
|
||||
func (s *Server) Stop(ctx context.Context) error {
|
||||
if s.httpServer != nil {
|
||||
if err := s.httpServer.Shutdown(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
s.opampServer.Stop()
|
||||
|
||||
// stop usage manager
|
||||
|
||||
@@ -108,7 +108,6 @@ function LogsExplorerViewsContainer({
|
||||
const [page, setPage] = useState<number>(1);
|
||||
const [logs, setLogs] = useState<ILog[]>([]);
|
||||
const [requestData, setRequestData] = useState<Query | null>(null);
|
||||
const [queryId, setQueryId] = useState<string>(v4());
|
||||
const [listChartQuery, setListChartQuery] = useState<Query | null>(null);
|
||||
|
||||
const [orderBy, setOrderBy] = useState<string>('timestamp:desc');
|
||||
@@ -180,12 +179,7 @@ function LogsExplorerViewsContainer({
|
||||
},
|
||||
undefined,
|
||||
listQueryKeyRef,
|
||||
{
|
||||
...(!isEmpty(queryId) &&
|
||||
selectedPanelType !== PANEL_TYPES.LIST && {
|
||||
'X-SIGNOZ-QUERY-ID': queryId,
|
||||
}),
|
||||
},
|
||||
undefined,
|
||||
// custom selected time interval to prevent recalculating the start and end timestamps before fetching next pages
|
||||
'custom',
|
||||
);
|
||||
@@ -250,10 +244,6 @@ function LogsExplorerViewsContainer({
|
||||
setRequestData(newRequestData);
|
||||
}, [isLimit, logs, listQuery, pageSize, stagedQuery, getRequestData, page]);
|
||||
|
||||
useEffect(() => {
|
||||
setQueryId(v4());
|
||||
}, [data]);
|
||||
|
||||
const logEventCalledRef = useRef(false);
|
||||
useEffect(() => {
|
||||
if (!logEventCalledRef.current && !isUndefined(data?.payload)) {
|
||||
|
||||
3
go.mod
3
go.mod
@@ -31,7 +31,6 @@ require (
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/gorilla/handlers v1.5.1
|
||||
github.com/gorilla/mux v1.8.1
|
||||
github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674
|
||||
github.com/huandu/go-sqlbuilder v1.39.1
|
||||
github.com/jackc/pgx/v5 v5.9.2
|
||||
github.com/json-iterator/go v1.1.13-0.20220915233716-71ac16282d12
|
||||
@@ -58,6 +57,7 @@ require (
|
||||
github.com/segmentio/analytics-go/v3 v3.2.1
|
||||
github.com/sethvargo/go-password v0.2.0
|
||||
github.com/smartystreets/goconvey v1.8.1
|
||||
github.com/soheilhy/cmux v0.1.5
|
||||
github.com/spf13/cobra v1.10.2
|
||||
github.com/stretchr/testify v1.11.1
|
||||
github.com/swaggest/jsonschema-go v0.3.78
|
||||
@@ -133,6 +133,7 @@ require (
|
||||
github.com/go-playground/universal-translator v0.18.1 // indirect
|
||||
github.com/goccy/go-yaml v1.19.2 // indirect
|
||||
github.com/google/go-cmp v0.7.0 // indirect
|
||||
github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect
|
||||
github.com/hashicorp/go-metrics v0.5.4 // indirect
|
||||
github.com/huandu/go-clone v1.7.3 // indirect
|
||||
github.com/leodido/go-urn v1.4.0 // indirect
|
||||
|
||||
3
go.sum
3
go.sum
@@ -1057,6 +1057,8 @@ github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1
|
||||
github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
|
||||
github.com/smartystreets/goconvey v1.8.1 h1:qGjIddxOk4grTu9JPOU31tVfq3cNdBlNa5sSznIX1xY=
|
||||
github.com/smartystreets/goconvey v1.8.1/go.mod h1:+/u4qLyY6x1jReYOp7GOM2FSt8aP9CzCZL03bI28W60=
|
||||
github.com/soheilhy/cmux v0.1.5 h1:jjzc5WVemNEDTLwv9tlmemhC73tI08BNOIGwBOo10Js=
|
||||
github.com/soheilhy/cmux v0.1.5/go.mod h1:T7TcVDs9LWfQgPlPsdngu6I6QIoyIFZDDC6sNE1GqG0=
|
||||
github.com/sourcegraph/conc v0.3.0 h1:OQTbbt6P72L20UqAkXXuLOj79LfEanQ+YQFNpLA9ySo=
|
||||
github.com/sourcegraph/conc v0.3.0/go.mod h1:Sdozi7LEKbFPqYX2/J+iBAM6HpqSLTASQIKqDmF7Mt0=
|
||||
github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
|
||||
@@ -1491,6 +1493,7 @@ golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81R
|
||||
golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
|
||||
golang.org/x/net v0.0.0-20201031054903-ff519b6c9102/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
|
||||
golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
|
||||
golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
|
||||
golang.org/x/net v0.0.0-20201209123823-ac852fbbde11/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
|
||||
golang.org/x/net v0.0.0-20201224014010-6772e930b67b/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
|
||||
golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
|
||||
|
||||
@@ -1,14 +1,10 @@
|
||||
package apiserver
|
||||
|
||||
import (
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
"github.com/gorilla/mux"
|
||||
)
|
||||
|
||||
type APIServer interface {
|
||||
// APIServer is a long running service serving the SigNoz API.
|
||||
factory.ServiceWithHealthy
|
||||
|
||||
// Returns the mux router for the API server. Primarily used for collecting OpenAPI operations.
|
||||
Router() *mux.Router
|
||||
|
||||
|
||||
@@ -8,8 +8,6 @@ import (
|
||||
|
||||
// Config holds the configuration for config.
|
||||
type Config struct {
|
||||
// Address is the TCP address the API server listens on, in the form "host:port".
|
||||
Address string `mapstructure:"address"`
|
||||
Timeout Timeout `mapstructure:"timeout"`
|
||||
Logging Logging `mapstructure:"logging"`
|
||||
}
|
||||
@@ -34,7 +32,6 @@ func NewConfigFactory() factory.ConfigFactory {
|
||||
|
||||
func newConfig() factory.Config {
|
||||
return &Config{
|
||||
Address: "0.0.0.0:8080",
|
||||
Timeout: Timeout{
|
||||
Default: 60 * time.Second,
|
||||
Max: 600 * time.Second,
|
||||
|
||||
@@ -13,7 +13,6 @@ import (
|
||||
)
|
||||
|
||||
func TestNewWithEnvProvider(t *testing.T) {
|
||||
t.Setenv("SIGNOZ_APISERVER_ADDRESS", "0.0.0.0:9090")
|
||||
t.Setenv("SIGNOZ_APISERVER_TIMEOUT_DEFAULT", "70s")
|
||||
t.Setenv("SIGNOZ_APISERVER_TIMEOUT_MAX", "700s")
|
||||
t.Setenv("SIGNOZ_APISERVER_TIMEOUT_EXCLUDED__ROUTES", "/excluded1,/excluded2")
|
||||
@@ -39,7 +38,6 @@ func TestNewWithEnvProvider(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
|
||||
expected := &Config{
|
||||
Address: "0.0.0.0:9090",
|
||||
Timeout: Timeout{
|
||||
Default: 70 * time.Second,
|
||||
Max: 700 * time.Second,
|
||||
|
||||
@@ -2,21 +2,16 @@ package signozapiserver
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/alertmanager"
|
||||
"github.com/SigNoz/signoz/pkg/apiserver"
|
||||
"github.com/SigNoz/signoz/pkg/auditor"
|
||||
"github.com/SigNoz/signoz/pkg/authz"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
"github.com/SigNoz/signoz/pkg/flagger"
|
||||
"github.com/SigNoz/signoz/pkg/gateway"
|
||||
"github.com/SigNoz/signoz/pkg/global"
|
||||
"github.com/SigNoz/signoz/pkg/http/handler"
|
||||
"github.com/SigNoz/signoz/pkg/http/middleware"
|
||||
httpserver "github.com/SigNoz/signoz/pkg/http/server"
|
||||
"github.com/SigNoz/signoz/pkg/identn"
|
||||
"github.com/SigNoz/signoz/pkg/licensing"
|
||||
"github.com/SigNoz/signoz/pkg/modules/aiobservability"
|
||||
"github.com/SigNoz/signoz/pkg/modules/authdomain"
|
||||
@@ -42,22 +37,18 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
"github.com/SigNoz/signoz/pkg/querier"
|
||||
"github.com/SigNoz/signoz/pkg/ruler"
|
||||
"github.com/SigNoz/signoz/pkg/sharder"
|
||||
"github.com/SigNoz/signoz/pkg/statsreporter"
|
||||
"github.com/SigNoz/signoz/pkg/subscription"
|
||||
"github.com/SigNoz/signoz/pkg/types"
|
||||
"github.com/SigNoz/signoz/pkg/types/authtypes"
|
||||
"github.com/SigNoz/signoz/pkg/web"
|
||||
"github.com/SigNoz/signoz/pkg/zeus"
|
||||
"github.com/gorilla/mux"
|
||||
)
|
||||
|
||||
type provider struct {
|
||||
globalConfig global.Config
|
||||
web web.Web
|
||||
config apiserver.Config
|
||||
settings factory.ScopedProviderSettings
|
||||
router *mux.Router
|
||||
httpServer *httpserver.Server
|
||||
healthyC chan struct{}
|
||||
authzMiddleware *middleware.AuthZ
|
||||
authzService authz.AuthZ
|
||||
orgHandler organization.Handler
|
||||
@@ -141,11 +132,6 @@ func NewFactory(
|
||||
rulerHandler ruler.Handler,
|
||||
statsHandler statsreporter.Handler,
|
||||
savedViewHandler savedview.Handler,
|
||||
globalConfig global.Config,
|
||||
identNResolver identn.IdentNResolver,
|
||||
sharder sharder.Sharder,
|
||||
auditor auditor.Auditor,
|
||||
web web.Web,
|
||||
quickFilterModule quickfilter.Module,
|
||||
quickFilterHandler quickfilter.Handler,
|
||||
) factory.ProviderFactory[apiserver.APIServer, apiserver.Config] {
|
||||
@@ -193,11 +179,6 @@ func NewFactory(
|
||||
rulerHandler,
|
||||
statsHandler,
|
||||
savedViewHandler,
|
||||
globalConfig,
|
||||
identNResolver,
|
||||
sharder,
|
||||
auditor,
|
||||
web,
|
||||
quickFilterModule,
|
||||
quickFilterHandler,
|
||||
)
|
||||
@@ -247,11 +228,6 @@ func newProvider(
|
||||
rulerHandler ruler.Handler,
|
||||
statsHandler statsreporter.Handler,
|
||||
savedViewHandler savedview.Handler,
|
||||
globalConfig global.Config,
|
||||
identNResolver identn.IdentNResolver,
|
||||
sharder sharder.Sharder,
|
||||
auditor auditor.Auditor,
|
||||
web web.Web,
|
||||
quickFilterModule quickfilter.Module,
|
||||
quickFilterHandler quickfilter.Handler,
|
||||
) (apiserver.APIServer, error) {
|
||||
@@ -259,10 +235,9 @@ func newProvider(
|
||||
router := mux.NewRouter().UseEncodedPath()
|
||||
|
||||
provider := &provider{
|
||||
globalConfig: globalConfig,
|
||||
web: web,
|
||||
config: config,
|
||||
settings: settings,
|
||||
router: router,
|
||||
healthyC: make(chan struct{}),
|
||||
orgHandler: orgHandler,
|
||||
userHandler: userHandler,
|
||||
authzService: authzService,
|
||||
@@ -307,72 +282,13 @@ func newProvider(
|
||||
|
||||
provider.authzMiddleware = middleware.NewAuthZ(settings.Logger(), orgGetter, authzService)
|
||||
|
||||
router.Use(middleware.NewRecovery(settings.Logger()).Wrap)
|
||||
router.Use(middleware.NewOtel("apiserver", providerSettings.MeterProvider, providerSettings.TracerProvider).Wrap)
|
||||
router.Use(middleware.NewIdentN(identNResolver, sharder, settings.Logger()).Wrap)
|
||||
router.Use(middleware.NewTimeout(settings.Logger(),
|
||||
config.Timeout.ExcludedRoutes,
|
||||
config.Timeout.Default,
|
||||
config.Timeout.Max,
|
||||
).Wrap)
|
||||
router.Use(middleware.NewResource(settings.Logger()).Wrap)
|
||||
router.Use(middleware.NewAudit(settings.Logger(), config.Logging.ExcludedRoutes, auditor).Wrap)
|
||||
router.Use(middleware.NewComment().Wrap)
|
||||
|
||||
if err := provider.AddToRouter(router); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
httpHandler := middleware.NewCors().Wrap(router)
|
||||
httpHandler = middleware.NewCompress().Wrap(httpHandler)
|
||||
|
||||
routePrefix := globalConfig.ExternalPath()
|
||||
if routePrefix != "" {
|
||||
prefixed := http.StripPrefix(routePrefix, httpHandler)
|
||||
httpHandler = http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
|
||||
switch req.URL.Path {
|
||||
case "/api/v1/health", "/api/v2/healthz", "/api/v2/readyz", "/api/v2/livez":
|
||||
router.ServeHTTP(w, req)
|
||||
return
|
||||
}
|
||||
|
||||
prefixed.ServeHTTP(w, req)
|
||||
})
|
||||
}
|
||||
|
||||
if config.Address == "" {
|
||||
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "apiserver.address is required")
|
||||
}
|
||||
|
||||
httpServer, err := httpserver.New(settings.Logger(), httpserver.Config{Address: config.Address}, httpHandler)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
provider.httpServer = httpServer
|
||||
|
||||
return provider, nil
|
||||
}
|
||||
|
||||
func (provider *provider) Start(ctx context.Context) error {
|
||||
// Mount the web routes last so the catch-all prefix does not shadow API
|
||||
// routes registered on the router after construction.
|
||||
if err := provider.web.AddToRouter(provider.router); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
close(provider.healthyC)
|
||||
|
||||
return provider.httpServer.Start(ctx)
|
||||
}
|
||||
|
||||
func (provider *provider) Stop(ctx context.Context) error {
|
||||
return provider.httpServer.Stop(ctx)
|
||||
}
|
||||
|
||||
func (provider *provider) Healthy() <-chan struct{} {
|
||||
return provider.healthyC
|
||||
}
|
||||
|
||||
func (provider *provider) Router() *mux.Router {
|
||||
return provider.router
|
||||
}
|
||||
|
||||
@@ -78,40 +78,6 @@ func NewRegistry(ctx context.Context, logger *slog.Logger, services ...NamedServ
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Add registers additional services into the registry. It must be called before Start.
|
||||
func (registry *Registry) Add(ctx context.Context, services ...NamedService) error {
|
||||
added := make([]*serviceWithState, 0, len(services))
|
||||
for _, s := range services {
|
||||
if _, ok := registry.servicesByName[s.Name()]; ok {
|
||||
return errors.Newf(errors.TypeInvalidInput, ErrCodeInvalidRegistry, "cannot add service, duplicate service name %q", s.Name())
|
||||
}
|
||||
added = append(added, newServiceWithState(s))
|
||||
}
|
||||
|
||||
for _, ss := range added {
|
||||
registry.services = append(registry.services, ss)
|
||||
registry.servicesByName[ss.service.Name()] = ss
|
||||
}
|
||||
|
||||
for _, ss := range added {
|
||||
for _, dep := range ss.service.DependsOn() {
|
||||
if dep == ss.service.Name() {
|
||||
registry.logger.ErrorContext(ctx, "ignoring self-dependency", slog.Any("service", ss.service.Name()))
|
||||
continue
|
||||
}
|
||||
|
||||
if _, ok := registry.servicesByName[dep]; !ok {
|
||||
registry.logger.ErrorContext(ctx, "ignoring unknown dependency", slog.Any("service", ss.service.Name()), slog.Any("dependency", dep))
|
||||
continue
|
||||
}
|
||||
|
||||
ss.dependsOn = append(ss.dependsOn, dep)
|
||||
}
|
||||
}
|
||||
|
||||
return detectCyclicDeps(registry.services)
|
||||
}
|
||||
|
||||
func (registry *Registry) Start(ctx context.Context) {
|
||||
for _, ss := range registry.services {
|
||||
go func(ss *serviceWithState) {
|
||||
|
||||
@@ -342,61 +342,3 @@ func TestDependsOnCycleReturnsError(t *testing.T) {
|
||||
assert.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "dependency cycles detected")
|
||||
}
|
||||
|
||||
func TestRegistryAdd(t *testing.T) {
|
||||
s1 := newTestService(t)
|
||||
s2 := newTestService(t)
|
||||
|
||||
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1))
|
||||
require.NoError(t, err)
|
||||
|
||||
require.NoError(t, registry.Add(context.Background(), NewNamedService(MustNewName("s2"), s2)))
|
||||
|
||||
ctx := context.Background()
|
||||
registry.Start(ctx)
|
||||
|
||||
require.NoError(t, registry.AwaitHealthy(ctx))
|
||||
byState := registry.ServicesByState()
|
||||
assert.Len(t, byState[StateRunning], 2)
|
||||
assert.True(t, registry.IsHealthy())
|
||||
|
||||
assert.NoError(t, registry.Stop(ctx))
|
||||
}
|
||||
|
||||
func TestRegistryAddDuplicateReturnsError(t *testing.T) {
|
||||
s1 := newTestService(t)
|
||||
|
||||
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1))
|
||||
require.NoError(t, err)
|
||||
|
||||
err = registry.Add(context.Background(), NewNamedService(MustNewName("s1"), newTestService(t)))
|
||||
assert.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "duplicate service name")
|
||||
}
|
||||
|
||||
func TestRegistryAddWithDependency(t *testing.T) {
|
||||
s1 := newHealthyTestService(t)
|
||||
s2 := newTestService(t)
|
||||
|
||||
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1))
|
||||
require.NoError(t, err)
|
||||
|
||||
// s2 depends on the already registered s1.
|
||||
require.NoError(t, registry.Add(context.Background(), NewNamedService(MustNewName("s2"), s2, MustNewName("s1"))))
|
||||
|
||||
ctx := context.Background()
|
||||
registry.Start(ctx)
|
||||
|
||||
// s2 stays in STARTING until s1 is healthy.
|
||||
require.Eventually(t, func() bool {
|
||||
byState := registry.ServicesByState()
|
||||
return len(byState[StateStarting]) == 2
|
||||
}, time.Second, time.Millisecond)
|
||||
|
||||
close(s1.healthyC)
|
||||
|
||||
require.NoError(t, registry.AwaitHealthy(ctx))
|
||||
assert.True(t, registry.IsHealthy())
|
||||
|
||||
assert.NoError(t, registry.Stop(ctx))
|
||||
}
|
||||
|
||||
@@ -1,17 +0,0 @@
|
||||
package middleware
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
gorillahandlers "github.com/gorilla/handlers"
|
||||
)
|
||||
|
||||
type Compress struct{}
|
||||
|
||||
func NewCompress() *Compress {
|
||||
return &Compress{}
|
||||
}
|
||||
|
||||
func (middleware *Compress) Wrap(next http.Handler) http.Handler {
|
||||
return gorillahandlers.CompressHandler(next)
|
||||
}
|
||||
@@ -1,25 +0,0 @@
|
||||
package middleware
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/rs/cors"
|
||||
)
|
||||
|
||||
type Cors struct {
|
||||
cors *cors.Cors
|
||||
}
|
||||
|
||||
func NewCors() *Cors {
|
||||
return &Cors{
|
||||
cors: cors.New(cors.Options{
|
||||
AllowedOrigins: []string{"*"},
|
||||
AllowedMethods: []string{"GET", "DELETE", "POST", "PUT", "PATCH", "OPTIONS"},
|
||||
AllowedHeaders: []string{"Accept", "Authorization", "Content-Type", "cache-control", "X-SIGNOZ-QUERY-ID", "Sec-WebSocket-Protocol"},
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
func (middleware *Cors) Wrap(next http.Handler) http.Handler {
|
||||
return middleware.cors.Handler(next)
|
||||
}
|
||||
@@ -1,43 +0,0 @@
|
||||
package middleware
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"slices"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"go.opentelemetry.io/contrib/instrumentation/github.com/gorilla/mux/otelmux"
|
||||
"go.opentelemetry.io/otel/metric"
|
||||
"go.opentelemetry.io/otel/propagation"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
)
|
||||
|
||||
// defaultExcludedRoutes are the health endpoints kept out of tracing/metrics to
|
||||
// avoid drowning telemetry in probe traffic.
|
||||
var defaultExcludedRoutes = []string{
|
||||
"/api/v1/health",
|
||||
"/api/v2/healthz",
|
||||
"/api/v2/readyz",
|
||||
"/api/v2/livez",
|
||||
}
|
||||
|
||||
type Otel struct {
|
||||
wrap mux.MiddlewareFunc
|
||||
}
|
||||
|
||||
func NewOtel(service string, meterProvider metric.MeterProvider, tracerProvider trace.TracerProvider) *Otel {
|
||||
return &Otel{
|
||||
wrap: otelmux.Middleware(
|
||||
service,
|
||||
otelmux.WithMeterProvider(meterProvider),
|
||||
otelmux.WithTracerProvider(tracerProvider),
|
||||
otelmux.WithPropagators(propagation.NewCompositeTextMapPropagator(propagation.Baggage{}, propagation.TraceContext{})),
|
||||
otelmux.WithFilter(func(r *http.Request) bool {
|
||||
return !slices.Contains(defaultExcludedRoutes, r.URL.Path)
|
||||
}),
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
func (middleware *Otel) Wrap(next http.Handler) http.Handler {
|
||||
return middleware.wrap(next)
|
||||
}
|
||||
@@ -1,14 +1,9 @@
|
||||
package server
|
||||
|
||||
import "time"
|
||||
|
||||
// Config holds the configuration for http.
|
||||
type Config struct {
|
||||
//Address specifies the TCP address for the server to listen on, in the form "host:port".
|
||||
// If empty, ":http" (port 80) is used. The service names are defined in RFC 6335 and assigned by IANA.
|
||||
// See net.Dial for details of the address format.
|
||||
Address string `mapstructure:"address"`
|
||||
|
||||
// Timeout bounds both reading the request and writing the response. Zero means no timeout.
|
||||
Timeout time.Duration `mapstructure:"timeout"`
|
||||
}
|
||||
|
||||
@@ -31,8 +31,8 @@ func New(logger *slog.Logger, cfg Config, handler http.Handler) (*Server, error)
|
||||
srv := &http.Server{
|
||||
Addr: cfg.Address,
|
||||
Handler: handler,
|
||||
ReadTimeout: cfg.Timeout,
|
||||
WriteTimeout: cfg.Timeout,
|
||||
ReadTimeout: 10 * time.Second,
|
||||
WriteTimeout: 10 * time.Second,
|
||||
MaxHeaderBytes: 1 << 20,
|
||||
}
|
||||
|
||||
|
||||
@@ -45,7 +45,7 @@ func newConfig() factory.Config {
|
||||
return &Config{
|
||||
Tokenizer: TokenizerConfig{
|
||||
Enabled: true,
|
||||
Headers: []string{"Authorization", "Sec-WebSocket-Protocol"},
|
||||
Headers: []string{"Authorization"},
|
||||
},
|
||||
APIKeyConfig: APIKeyConfig{
|
||||
Enabled: true,
|
||||
|
||||
@@ -6,7 +6,6 @@ import (
|
||||
"net/http"
|
||||
nethttppprof "net/http/pprof"
|
||||
runtimepprof "runtime/pprof"
|
||||
"time"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
httpserver "github.com/SigNoz/signoz/pkg/http/server"
|
||||
@@ -24,7 +23,7 @@ func NewFactory() factory.ProviderFactory[pprof.PProf, pprof.Config] {
|
||||
func New(_ context.Context, settings factory.ProviderSettings, config pprof.Config) (pprof.PProf, error) {
|
||||
server, err := httpserver.New(
|
||||
settings.Logger.With(slog.String("pkg", "github.com/SigNoz/signoz/pkg/pprof/httppprof")),
|
||||
httpserver.Config{Address: config.Address, Timeout: 10 * time.Second},
|
||||
httpserver.Config{Address: config.Address},
|
||||
newHandler(),
|
||||
)
|
||||
if err != nil {
|
||||
|
||||
@@ -1,243 +0,0 @@
|
||||
package queryprogress
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/model"
|
||||
"github.com/google/uuid"
|
||||
"golang.org/x/exp/maps"
|
||||
)
|
||||
|
||||
// tracks progress and manages subscriptions for all queries
|
||||
type inMemoryQueryProgressTracker struct {
|
||||
logger *slog.Logger
|
||||
queries map[string]*queryTracker
|
||||
lock sync.RWMutex
|
||||
}
|
||||
|
||||
func (tracker *inMemoryQueryProgressTracker) ReportQueryStarted(
|
||||
queryId string,
|
||||
) (postQueryCleanup func(), apiErr *model.ApiError) {
|
||||
tracker.lock.Lock()
|
||||
defer tracker.lock.Unlock()
|
||||
|
||||
_, exists := tracker.queries[queryId]
|
||||
if exists {
|
||||
return nil, model.BadRequest(fmt.Errorf(
|
||||
"query %s already started", queryId,
|
||||
))
|
||||
}
|
||||
|
||||
tracker.queries[queryId] = newQueryTracker(tracker.logger, queryId)
|
||||
|
||||
return func() {
|
||||
tracker.onQueryFinished(queryId)
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (tracker *inMemoryQueryProgressTracker) ReportQueryProgress(
|
||||
queryId string, chProgress *clickhouse.Progress,
|
||||
) *model.ApiError {
|
||||
queryTracker, err := tracker.getQueryTracker(queryId)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
queryTracker.handleProgressUpdate(chProgress)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (tracker *inMemoryQueryProgressTracker) SubscribeToQueryProgress(
|
||||
queryId string,
|
||||
) (<-chan model.QueryProgress, func(), *model.ApiError) {
|
||||
queryTracker, err := tracker.getQueryTracker(queryId)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
return queryTracker.subscribe()
|
||||
}
|
||||
|
||||
func (tracker *inMemoryQueryProgressTracker) onQueryFinished(
|
||||
queryId string,
|
||||
) {
|
||||
tracker.lock.Lock()
|
||||
queryTracker := tracker.queries[queryId]
|
||||
if queryTracker != nil {
|
||||
delete(tracker.queries, queryId)
|
||||
}
|
||||
tracker.lock.Unlock()
|
||||
|
||||
if queryTracker != nil {
|
||||
queryTracker.onFinished()
|
||||
}
|
||||
}
|
||||
|
||||
func (tracker *inMemoryQueryProgressTracker) getQueryTracker(
|
||||
queryId string,
|
||||
) (*queryTracker, *model.ApiError) {
|
||||
tracker.lock.RLock()
|
||||
defer tracker.lock.RUnlock()
|
||||
|
||||
queryTracker := tracker.queries[queryId]
|
||||
if queryTracker == nil {
|
||||
return nil, model.NotFoundError(fmt.Errorf(
|
||||
"query %s doesn't exist", queryId,
|
||||
))
|
||||
}
|
||||
|
||||
return queryTracker, nil
|
||||
}
|
||||
|
||||
// Tracks progress and manages subscriptions for a single query
|
||||
type queryTracker struct {
|
||||
logger *slog.Logger
|
||||
queryId string
|
||||
isFinished bool
|
||||
|
||||
progress *model.QueryProgress
|
||||
subscriptions map[string]*queryProgressSubscription
|
||||
|
||||
lock sync.Mutex
|
||||
}
|
||||
|
||||
func newQueryTracker(logger *slog.Logger, queryId string) *queryTracker {
|
||||
return &queryTracker{
|
||||
logger: logger,
|
||||
queryId: queryId,
|
||||
subscriptions: map[string]*queryProgressSubscription{},
|
||||
}
|
||||
}
|
||||
|
||||
func (qt *queryTracker) handleProgressUpdate(p *clickhouse.Progress) {
|
||||
qt.lock.Lock()
|
||||
defer qt.lock.Unlock()
|
||||
|
||||
if qt.isFinished {
|
||||
qt.logger.Warn("received clickhouse progress update for finished query", "queryId", qt.queryId, "progress", p)
|
||||
return
|
||||
}
|
||||
|
||||
if qt.progress == nil {
|
||||
// This is the first update
|
||||
qt.progress = &model.QueryProgress{}
|
||||
}
|
||||
updateQueryProgress(qt.progress, p)
|
||||
|
||||
// broadcast latest state to all subscribers.
|
||||
for _, sub := range maps.Values(qt.subscriptions) {
|
||||
sub.send(*qt.progress)
|
||||
}
|
||||
}
|
||||
|
||||
func (qt *queryTracker) subscribe() (
|
||||
<-chan model.QueryProgress, func(), *model.ApiError,
|
||||
) {
|
||||
qt.lock.Lock()
|
||||
defer qt.lock.Unlock()
|
||||
|
||||
if qt.isFinished {
|
||||
return nil, nil, model.NotFoundError(fmt.Errorf(
|
||||
"query %s already finished", qt.queryId,
|
||||
))
|
||||
}
|
||||
|
||||
subscriberId := uuid.NewString()
|
||||
subscription := newQueryProgressSubscription(qt.logger)
|
||||
qt.subscriptions[subscriberId] = subscription
|
||||
|
||||
if qt.progress != nil {
|
||||
subscription.send(*qt.progress)
|
||||
}
|
||||
|
||||
return subscription.ch, func() {
|
||||
qt.unsubscribe(subscriberId)
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (qt *queryTracker) unsubscribe(subscriberId string) {
|
||||
qt.lock.Lock()
|
||||
defer qt.lock.Unlock()
|
||||
|
||||
if qt.isFinished {
|
||||
qt.logger.Debug("received unsubscribe request after query finished", "subscriber", subscriberId, "queryId", qt.queryId)
|
||||
return
|
||||
}
|
||||
|
||||
subscription := qt.subscriptions[subscriberId]
|
||||
if subscription != nil {
|
||||
subscription.close()
|
||||
delete(qt.subscriptions, subscriberId)
|
||||
}
|
||||
}
|
||||
|
||||
func (qt *queryTracker) onFinished() {
|
||||
qt.lock.Lock()
|
||||
defer qt.lock.Unlock()
|
||||
|
||||
if qt.isFinished {
|
||||
qt.logger.Warn("receiver query finish report after query finished", "queryId", qt.queryId)
|
||||
return
|
||||
}
|
||||
|
||||
for subId, sub := range qt.subscriptions {
|
||||
sub.close()
|
||||
delete(qt.subscriptions, subId)
|
||||
}
|
||||
|
||||
qt.isFinished = true
|
||||
}
|
||||
|
||||
type queryProgressSubscription struct {
|
||||
logger *slog.Logger
|
||||
ch chan model.QueryProgress
|
||||
isClosed bool
|
||||
lock sync.Mutex
|
||||
}
|
||||
|
||||
func newQueryProgressSubscription(logger *slog.Logger) *queryProgressSubscription {
|
||||
ch := make(chan model.QueryProgress, 1000)
|
||||
return &queryProgressSubscription{
|
||||
logger: logger,
|
||||
ch: ch,
|
||||
}
|
||||
}
|
||||
|
||||
// Must not block or panic in any scenario
|
||||
func (ch *queryProgressSubscription) send(progress model.QueryProgress) {
|
||||
ch.lock.Lock()
|
||||
defer ch.lock.Unlock()
|
||||
|
||||
if ch.isClosed {
|
||||
ch.logger.Error("can't send query progress: channel already closed.", "progress", progress)
|
||||
return
|
||||
}
|
||||
|
||||
// subscription channels are expected to have big enough buffers to ensure
|
||||
// blocking while sending doesn't happen in the happy path
|
||||
select {
|
||||
case ch.ch <- progress:
|
||||
ch.logger.Debug("published query progress", "progress", progress)
|
||||
default:
|
||||
ch.logger.Error("couldn't publish query progress. dropping update.", "progress", progress)
|
||||
}
|
||||
}
|
||||
|
||||
func (ch *queryProgressSubscription) close() {
|
||||
ch.lock.Lock()
|
||||
defer ch.lock.Unlock()
|
||||
|
||||
if !ch.isClosed {
|
||||
close(ch.ch)
|
||||
ch.isClosed = true
|
||||
}
|
||||
}
|
||||
|
||||
func updateQueryProgress(qp *model.QueryProgress, chProgress *clickhouse.Progress) {
|
||||
qp.ReadRows += chProgress.Rows
|
||||
qp.ReadBytes += chProgress.Bytes
|
||||
qp.ElapsedMs += uint64(chProgress.Elapsed.Milliseconds())
|
||||
}
|
||||
@@ -1,33 +0,0 @@
|
||||
package queryprogress
|
||||
|
||||
import (
|
||||
"log/slog"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/model"
|
||||
)
|
||||
|
||||
type QueryProgressTracker interface {
|
||||
// Tells the tracker that query with id `queryId` has started.
|
||||
// Progress can only be reported for and tracked for a query that is in progress.
|
||||
// Returns a cleanup function that must be called after the query finishes.
|
||||
ReportQueryStarted(queryId string) (postQueryCleanup func(), apiErr *model.ApiError)
|
||||
|
||||
// Report progress stats received from clickhouse for `queryId`
|
||||
ReportQueryProgress(queryId string, chProgress *clickhouse.Progress) *model.ApiError
|
||||
|
||||
// Subscribe to progress updates for `queryId`
|
||||
// The returned channel will produce `QueryProgress` instances representing
|
||||
// the latest state of query progress stats. Also returns a function that
|
||||
// can be called to unsubscribe before the query finishes, if needed.
|
||||
SubscribeToQueryProgress(queryId string) (ch <-chan model.QueryProgress, unsubscribe func(), apiErr *model.ApiError)
|
||||
}
|
||||
|
||||
func NewQueryProgressTracker(logger *slog.Logger) QueryProgressTracker {
|
||||
// InMemory tracker is useful only for single replica query service setups.
|
||||
// Multi replica setups must use a centralized store for tracking and subscribing to query progress
|
||||
return &inMemoryQueryProgressTracker{
|
||||
logger: logger,
|
||||
queries: map[string]*queryTracker{},
|
||||
}
|
||||
}
|
||||
@@ -1,102 +0,0 @@
|
||||
package queryprogress
|
||||
|
||||
import (
|
||||
"log/slog"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/model"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestQueryProgressTracking(t *testing.T) {
|
||||
require := require.New(t)
|
||||
|
||||
tracker := NewQueryProgressTracker(slog.Default())
|
||||
|
||||
testQueryId := "test-query"
|
||||
|
||||
testProgress := &clickhouse.Progress{}
|
||||
err := tracker.ReportQueryProgress(testQueryId, testProgress)
|
||||
require.NotNil(err, "shouldn't be able to report query progress before query has been started")
|
||||
require.Equal(err.Type(), model.ErrorNotFound)
|
||||
|
||||
ch, unsubscribe, err := tracker.SubscribeToQueryProgress(testQueryId)
|
||||
require.NotNil(err, "shouldn't be able to subscribe for progress updates before query has been started")
|
||||
require.Equal(err.Type(), model.ErrorNotFound)
|
||||
require.Nil(ch)
|
||||
require.Nil(unsubscribe)
|
||||
|
||||
reportQueryFinished, err := tracker.ReportQueryStarted(testQueryId)
|
||||
require.Nil(err, "should be able to report start of a query to be tracked")
|
||||
|
||||
testProgress1 := &clickhouse.Progress{
|
||||
Rows: 10,
|
||||
Bytes: 20,
|
||||
TotalRows: 100,
|
||||
Elapsed: 20 * time.Millisecond,
|
||||
}
|
||||
err = tracker.ReportQueryProgress(testQueryId, testProgress1)
|
||||
require.Nil(err, "should be able to report progress after query has started")
|
||||
|
||||
ch, unsubscribe, err = tracker.SubscribeToQueryProgress(testQueryId)
|
||||
require.Nil(err, "should be able to subscribe to query progress updates after query started")
|
||||
require.NotNil(ch)
|
||||
require.NotNil(unsubscribe)
|
||||
|
||||
expectedProgress := model.QueryProgress{}
|
||||
updateQueryProgress(&expectedProgress, testProgress1)
|
||||
require.Equal(expectedProgress.ReadRows, testProgress1.Rows)
|
||||
select {
|
||||
case qp := <-ch:
|
||||
require.Equal(qp, expectedProgress)
|
||||
default:
|
||||
require.Fail("should receive latest query progress state immediately after subscription")
|
||||
}
|
||||
select {
|
||||
case _ = <-ch:
|
||||
require.Fail("should have had only one pending update at this point")
|
||||
default:
|
||||
}
|
||||
|
||||
testProgress2 := &clickhouse.Progress{
|
||||
Rows: 20,
|
||||
Bytes: 40,
|
||||
TotalRows: 100,
|
||||
Elapsed: 40 * time.Millisecond,
|
||||
}
|
||||
err = tracker.ReportQueryProgress(testQueryId, testProgress2)
|
||||
require.Nil(err, "should be able to report progress multiple times while query is in progress")
|
||||
|
||||
updateQueryProgress(&expectedProgress, testProgress2)
|
||||
select {
|
||||
case qp := <-ch:
|
||||
require.Equal(qp, expectedProgress)
|
||||
default:
|
||||
require.Fail("should receive updates whenever new progress updates get reported to tracker")
|
||||
}
|
||||
select {
|
||||
case _ = <-ch:
|
||||
require.Fail("should have had only one pending update at this point")
|
||||
default:
|
||||
}
|
||||
|
||||
reportQueryFinished()
|
||||
select {
|
||||
case _, isSubscriptionChannelOpen := <-ch:
|
||||
require.False(isSubscriptionChannelOpen, "subscription channels should get closed after query finishes")
|
||||
default:
|
||||
require.Fail("subscription channels should get closed after query finishes")
|
||||
}
|
||||
|
||||
err = tracker.ReportQueryProgress(testQueryId, testProgress)
|
||||
require.NotNil(err, "shouldn't be able to report query progress after query has finished")
|
||||
require.Equal(err.Type(), model.ErrorNotFound)
|
||||
|
||||
ch, unsubscribe, err = tracker.SubscribeToQueryProgress(testQueryId)
|
||||
require.NotNil(err, "shouldn't be able to subscribe for progress updates after query has finished")
|
||||
require.Equal(err.Type(), model.ErrorNotFound)
|
||||
require.Nil(ch)
|
||||
require.Nil(unsubscribe)
|
||||
}
|
||||
@@ -44,7 +44,6 @@ import (
|
||||
|
||||
"log/slog"
|
||||
|
||||
queryprogress "github.com/SigNoz/signoz/pkg/query-service/app/clickhouseReader/query_progress"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/app/resource"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/app/services"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/app/traces/smart"
|
||||
@@ -145,7 +144,6 @@ type ClickHouseReader struct {
|
||||
logsResourceKeys string
|
||||
logsTagAttributeTableV2 string
|
||||
logger *slog.Logger
|
||||
queryProgressTracker queryprogress.QueryProgressTracker
|
||||
|
||||
logsTableV2 string
|
||||
logsLocalTableV2 string
|
||||
@@ -214,7 +212,6 @@ func NewReader(
|
||||
logsTagAttributeTableV2: options.primary.LogsTagAttributeTableV2,
|
||||
liveTailRefreshSeconds: options.primary.LiveTailRefreshSeconds,
|
||||
cluster: cluster,
|
||||
queryProgressTracker: queryprogress.NewQueryProgressTracker(logger),
|
||||
logsTableV2: options.primary.LogsTableV2,
|
||||
logsLocalTableV2: options.primary.LogsLocalTableV2,
|
||||
logsResourceTableV2: options.primary.LogsResourceTableV2,
|
||||
@@ -4024,27 +4021,6 @@ func (r *ClickHouseReader) GetTimeSeriesResultV3(ctx context.Context, query stri
|
||||
instrumentationtypes.CodeNamespace: "clickhouse-reader",
|
||||
instrumentationtypes.CodeFunctionName: "GetTimeSeriesResultV3",
|
||||
})
|
||||
// Hook up query progress reporting if requested.
|
||||
queryId := ctx.Value("queryId")
|
||||
if queryId != nil {
|
||||
qid, ok := queryId.(string)
|
||||
if !ok {
|
||||
r.logger.Error("GetTimeSeriesResultV3: queryId in ctx not a string as expected", "queryId", queryId)
|
||||
|
||||
} else {
|
||||
ctx = clickhouse.Context(ctx, clickhouse.WithProgress(
|
||||
func(p *clickhouse.Progress) {
|
||||
go func() {
|
||||
err := r.queryProgressTracker.ReportQueryProgress(qid, p)
|
||||
if err != nil {
|
||||
r.logger.Error("Couldn't report query progress", "queryId", qid, errorsV2.Attr(err))
|
||||
}
|
||||
}()
|
||||
},
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
rows, err := r.db.Query(ctx, query)
|
||||
|
||||
if err != nil {
|
||||
@@ -5005,18 +4981,6 @@ func (r *ClickHouseReader) GetMinAndMaxTimestampForTraceID(ctx context.Context,
|
||||
return minTime.UnixNano(), maxTime.UnixNano(), nil
|
||||
}
|
||||
|
||||
func (r *ClickHouseReader) ReportQueryStartForProgressTracking(
|
||||
queryId string,
|
||||
) (func(), *model.ApiError) {
|
||||
return r.queryProgressTracker.ReportQueryStarted(queryId)
|
||||
}
|
||||
|
||||
func (r *ClickHouseReader) SubscribeToQueryProgress(
|
||||
queryId string,
|
||||
) (<-chan model.QueryProgress, func(), *model.ApiError) {
|
||||
return r.queryProgressTracker.SubscribeToQueryProgress(queryId)
|
||||
}
|
||||
|
||||
func (r *ClickHouseReader) UpdateMetricsMetadata(ctx context.Context, orgID valuer.UUID, req *model.UpdateMetricsMetadata) *model.ApiError {
|
||||
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
|
||||
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalMetrics.StringValue(),
|
||||
|
||||
@@ -37,7 +37,6 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/gorilla/websocket"
|
||||
jsoniter "github.com/json-iterator/go"
|
||||
_ "modernc.org/sqlite"
|
||||
|
||||
@@ -116,9 +115,6 @@ type APIHandler struct {
|
||||
// is registers.
|
||||
SetupCompleted bool
|
||||
|
||||
// Websocket connection upgrader
|
||||
Upgrader *websocket.Upgrader
|
||||
|
||||
LicensingAPI licensing.API
|
||||
|
||||
QueryParserAPI *queryparser.API
|
||||
@@ -213,12 +209,6 @@ func NewAPIHandler(opts APIHandlerOpts, config signoz.Config) (*APIHandler, erro
|
||||
aH.SetupCompleted = true
|
||||
}
|
||||
|
||||
aH.Upgrader = &websocket.Upgrader{
|
||||
CheckOrigin: func(r *http.Request) bool {
|
||||
return true
|
||||
},
|
||||
}
|
||||
|
||||
return aH, nil
|
||||
}
|
||||
|
||||
@@ -360,18 +350,10 @@ func (aH *APIHandler) RegisterQueryRangeV3Routes(router *mux.Router, am *middlew
|
||||
|
||||
subRouter.HandleFunc("/filter_suggestions", am.ViewAccess(aH.getQueryBuilderSuggestions)).Methods(http.MethodGet)
|
||||
|
||||
// TODO(Raj): Remove this handler after /ws based path has been completely rolled out.
|
||||
subRouter.HandleFunc("/query_progress", am.ViewAccess(aH.GetQueryProgressUpdates)).Methods(http.MethodGet)
|
||||
|
||||
// live logs
|
||||
subRouter.HandleFunc("/logs/livetail", am.ViewAccess(aH.Signoz.Handlers.QuerierHandler.QueryRawStream)).Methods(http.MethodGet)
|
||||
}
|
||||
|
||||
func (aH *APIHandler) RegisterWebSocketPaths(router *mux.Router, am *middleware.AuthZ) {
|
||||
subRouter := router.PathPrefix("/ws").Subrouter()
|
||||
subRouter.HandleFunc("/query_progress", am.ViewAccess(aH.GetQueryProgressUpdates)).Methods(http.MethodGet)
|
||||
}
|
||||
|
||||
func (aH *APIHandler) RegisterQueryRangeV4Routes(router *mux.Router, am *middleware.AuthZ) {
|
||||
subRouter := router.PathPrefix("/api/v4").Subrouter()
|
||||
subRouter.HandleFunc("/query_range", am.ViewAccess(aH.QueryRangeV4)).Methods(http.MethodPost)
|
||||
@@ -3551,27 +3533,6 @@ func (aH *APIHandler) queryRangeV3(ctx context.Context, queryRangeParams *v3.Que
|
||||
}
|
||||
}
|
||||
|
||||
// Hook up query progress tracking if requested
|
||||
queryIdHeader := r.Header.Get("X-SIGNOZ-QUERY-ID")
|
||||
if len(queryIdHeader) > 0 {
|
||||
onQueryFinished, apiErr := aH.reader.ReportQueryStartForProgressTracking(queryIdHeader)
|
||||
|
||||
if apiErr != nil {
|
||||
aH.logger.ErrorContext(ctx, "failed to report query start for progress tracking",
|
||||
"query_id", queryIdHeader, errors.Attr(apiErr),
|
||||
)
|
||||
|
||||
} else {
|
||||
// Adding queryId to the context signals clickhouse queries to report progress
|
||||
//lint:ignore SA1029 ignore for now
|
||||
ctx = context.WithValue(ctx, "queryId", queryIdHeader)
|
||||
|
||||
defer func() {
|
||||
go onQueryFinished()
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
|
||||
instrumentationtypes.CodeNamespace: "app",
|
||||
instrumentationtypes.CodeFunctionName: "QueryRange",
|
||||
@@ -3753,73 +3714,6 @@ func (aH *APIHandler) QueryRangeV3(w http.ResponseWriter, r *http.Request) {
|
||||
aH.queryRangeV3(r.Context(), queryRangeParams, w, r)
|
||||
}
|
||||
|
||||
func (aH *APIHandler) GetQueryProgressUpdates(w http.ResponseWriter, r *http.Request) {
|
||||
// Upgrade connection to websocket, sending back the requested protocol
|
||||
// value for sec-websocket-protocol
|
||||
//
|
||||
// Since js websocket API doesn't allow setting headers, this header is often
|
||||
// used for passing auth tokens. As per websocket spec the connection will only
|
||||
// succeed if the requested `Sec-Websocket-Protocol` is sent back as a header
|
||||
// in the upgrade response (signifying that the protocol is supported by the server).
|
||||
upgradeResponseHeaders := http.Header{}
|
||||
requestedProtocol := r.Header.Get("Sec-WebSocket-Protocol")
|
||||
if len(requestedProtocol) > 0 {
|
||||
upgradeResponseHeaders.Add("Sec-WebSocket-Protocol", requestedProtocol)
|
||||
}
|
||||
|
||||
c, err := aH.Upgrader.Upgrade(w, r, upgradeResponseHeaders)
|
||||
if err != nil {
|
||||
RespondError(w, model.InternalError(fmt.Errorf(
|
||||
"couldn't upgrade connection: %w", err,
|
||||
)), nil)
|
||||
return
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
// Websocket upgrade complete. Subscribe to query progress and send updates to client
|
||||
//
|
||||
// Note: we handle any subscription problems (queryId query param missing or query already complete etc)
|
||||
// after the websocket connection upgrade by closing the channel.
|
||||
// The other option would be to handle the errors before websocket upgrade by sending an
|
||||
// error response instead of the upgrade response, but that leads to a generic websocket
|
||||
// connection failure on the client.
|
||||
|
||||
queryId := r.URL.Query().Get("q")
|
||||
|
||||
progressCh, unsubscribe, apiErr := aH.reader.SubscribeToQueryProgress(queryId)
|
||||
if apiErr != nil {
|
||||
// Shouldn't happen unless query progress requested after query finished
|
||||
aH.logger.WarnContext(r.Context(), "failed to subscribe to query progress",
|
||||
"query_id", queryId, errors.Attr(apiErr),
|
||||
)
|
||||
return
|
||||
}
|
||||
defer func() { go unsubscribe() }()
|
||||
|
||||
for queryProgress := range progressCh {
|
||||
msg, err := json.Marshal(queryProgress)
|
||||
if err != nil {
|
||||
aH.logger.ErrorContext(r.Context(), "failed to serialize progress message",
|
||||
"query_id", queryId, "progress", queryProgress, errors.Attr(err),
|
||||
)
|
||||
continue
|
||||
}
|
||||
|
||||
err = c.WriteMessage(websocket.TextMessage, msg)
|
||||
if err != nil {
|
||||
aH.logger.ErrorContext(r.Context(), "failed to write progress message to websocket",
|
||||
"query_id", queryId, "msg", string(msg), errors.Attr(err),
|
||||
)
|
||||
break
|
||||
|
||||
} else {
|
||||
aH.logger.DebugContext(r.Context(), "wrote progress message to websocket",
|
||||
"query_id", queryId, "msg", string(msg),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (aH *APIHandler) getMetricMetadata(w http.ResponseWriter, r *http.Request) {
|
||||
claims, err := authtypes.ClaimsFromContext(r.Context())
|
||||
if err != nil {
|
||||
|
||||
@@ -2,10 +2,19 @@ package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"slices"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/queryparser"
|
||||
|
||||
"github.com/gorilla/handlers"
|
||||
|
||||
"github.com/rs/cors"
|
||||
"github.com/soheilhy/cmux"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/http/middleware"
|
||||
"github.com/SigNoz/signoz/pkg/licensing/nooplicensing"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/agentConf"
|
||||
@@ -15,15 +24,31 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/query-service/app/opamp"
|
||||
opAmpModel "github.com/SigNoz/signoz/pkg/query-service/app/opamp/model"
|
||||
"github.com/SigNoz/signoz/pkg/signoz"
|
||||
"github.com/SigNoz/signoz/pkg/web"
|
||||
|
||||
"log/slog"
|
||||
|
||||
"go.opentelemetry.io/contrib/instrumentation/github.com/gorilla/mux/otelmux"
|
||||
"go.opentelemetry.io/otel/propagation"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/query-service/constants"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/healthcheck"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/utils"
|
||||
)
|
||||
|
||||
// Server runs auxiliary servers (opamp) alongside the signoz apiserver
|
||||
// Server runs HTTP, Mux and a grpc server
|
||||
type Server struct {
|
||||
config signoz.Config
|
||||
signoz *signoz.SigNoz
|
||||
|
||||
// public http router
|
||||
httpConn net.Listener
|
||||
httpServer *http.Server
|
||||
httpHostPort string
|
||||
|
||||
opampServer *opamp.Server
|
||||
|
||||
unavailableChannel chan healthcheck.Status
|
||||
}
|
||||
|
||||
// NewServer creates and initializes Server
|
||||
@@ -67,21 +92,20 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Register the legacy query-service routes on the apiserver router. The
|
||||
// apiserver owns the HTTP server and applies the middleware chain at serve
|
||||
// time, so these routes get the same treatment as the apiserver routes.
|
||||
r := signoz.APIServer.Router()
|
||||
am := middleware.NewAuthZ(signoz.Instrumentation.Logger(), signoz.Modules.OrgGetter, signoz.Authz)
|
||||
s := &Server{
|
||||
config: config,
|
||||
signoz: signoz,
|
||||
httpHostPort: constants.HTTPHostPort,
|
||||
unavailableChannel: make(chan healthcheck.Status),
|
||||
}
|
||||
|
||||
apiHandler.RegisterRoutes(r, am)
|
||||
apiHandler.RegisterLogsRoutes(r, am)
|
||||
apiHandler.RegisterIntegrationRoutes(r, am)
|
||||
apiHandler.RegisterQueryRangeV3Routes(r, am)
|
||||
apiHandler.RegisterWebSocketPaths(r, am)
|
||||
apiHandler.RegisterQueryRangeV4Routes(r, am)
|
||||
apiHandler.RegisterMessagingQueuesRoutes(r, am)
|
||||
apiHandler.RegisterThirdPartyApiRoutes(r, am)
|
||||
apiHandler.RegisterTraceFunnelsRoutes(r, am)
|
||||
httpServer, err := s.createPublicServer(apiHandler, signoz.Web)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
s.httpServer = httpServer
|
||||
|
||||
opAmpModel.Init(signoz.SQLStore, signoz.Instrumentation.Logger(), signoz.Modules.OrgGetter)
|
||||
|
||||
@@ -99,8 +123,6 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
s := &Server{}
|
||||
|
||||
s.opampServer = opamp.InitializeServer(
|
||||
&opAmpModel.AllAgents,
|
||||
agentConfMgr,
|
||||
@@ -110,14 +132,133 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// Start starts the opamp websocket server. The HTTP API server is started by
|
||||
// the signoz registry.
|
||||
// HealthCheckStatus returns health check status channel a client can subscribe to
|
||||
func (s Server) HealthCheckStatus() chan healthcheck.Status {
|
||||
return s.unavailableChannel
|
||||
}
|
||||
|
||||
func (s *Server) createPublicServer(api *APIHandler, web web.Web) (*http.Server, error) {
|
||||
r := NewRouter()
|
||||
|
||||
r.Use(middleware.NewRecovery(s.signoz.Instrumentation.Logger()).Wrap)
|
||||
r.Use(otelmux.Middleware(
|
||||
"apiserver",
|
||||
otelmux.WithMeterProvider(s.signoz.Instrumentation.MeterProvider()),
|
||||
otelmux.WithTracerProvider(s.signoz.Instrumentation.TracerProvider()),
|
||||
otelmux.WithPropagators(propagation.NewCompositeTextMapPropagator(propagation.Baggage{}, propagation.TraceContext{})),
|
||||
otelmux.WithFilter(func(r *http.Request) bool {
|
||||
return !slices.Contains([]string{"/api/v1/health"}, r.URL.Path)
|
||||
}),
|
||||
))
|
||||
r.Use(middleware.NewIdentN(s.signoz.IdentNResolver, s.signoz.Sharder, s.signoz.Instrumentation.Logger()).Wrap)
|
||||
r.Use(middleware.NewTimeout(s.signoz.Instrumentation.Logger(),
|
||||
s.config.APIServer.Timeout.ExcludedRoutes,
|
||||
s.config.APIServer.Timeout.Default,
|
||||
s.config.APIServer.Timeout.Max,
|
||||
).Wrap)
|
||||
r.Use(middleware.NewResource(s.signoz.Instrumentation.Logger()).Wrap)
|
||||
r.Use(middleware.NewAudit(s.signoz.Instrumentation.Logger(), s.config.APIServer.Logging.ExcludedRoutes, s.signoz.Auditor).Wrap)
|
||||
r.Use(middleware.NewComment().Wrap)
|
||||
|
||||
am := middleware.NewAuthZ(s.signoz.Instrumentation.Logger(), s.signoz.Modules.OrgGetter, s.signoz.Authz)
|
||||
|
||||
api.RegisterRoutes(r, am)
|
||||
api.RegisterLogsRoutes(r, am)
|
||||
api.RegisterIntegrationRoutes(r, am)
|
||||
api.RegisterQueryRangeV3Routes(r, am)
|
||||
api.RegisterQueryRangeV4Routes(r, am)
|
||||
api.RegisterMessagingQueuesRoutes(r, am)
|
||||
api.RegisterThirdPartyApiRoutes(r, am)
|
||||
api.RegisterTraceFunnelsRoutes(r, am)
|
||||
|
||||
err := s.signoz.APIServer.AddToRouter(r)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
c := cors.New(cors.Options{
|
||||
AllowedOrigins: []string{"*"},
|
||||
AllowedMethods: []string{"GET", "DELETE", "POST", "PUT", "PATCH", "OPTIONS"},
|
||||
AllowedHeaders: []string{"Accept", "Authorization", "Content-Type", "cache-control"},
|
||||
})
|
||||
|
||||
handler := c.Handler(r)
|
||||
|
||||
handler = handlers.CompressHandler(handler)
|
||||
|
||||
err = web.AddToRouter(r)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
routePrefix := s.config.Global.ExternalPath()
|
||||
if routePrefix != "" {
|
||||
prefixed := http.StripPrefix(routePrefix, handler)
|
||||
handler = http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
|
||||
switch req.URL.Path {
|
||||
case "/api/v1/health", "/api/v2/healthz", "/api/v2/readyz", "/api/v2/livez":
|
||||
r.ServeHTTP(w, req)
|
||||
return
|
||||
}
|
||||
|
||||
prefixed.ServeHTTP(w, req)
|
||||
})
|
||||
}
|
||||
|
||||
return &http.Server{
|
||||
Handler: handler,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// initListeners initialises listeners of the server
|
||||
func (s *Server) initListeners() error {
|
||||
// listen on public port
|
||||
var err error
|
||||
publicHostPort := s.httpHostPort
|
||||
if publicHostPort == "" {
|
||||
return fmt.Errorf("constants.HTTPHostPort is required")
|
||||
}
|
||||
|
||||
s.httpConn, err = net.Listen("tcp", publicHostPort)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
slog.Info(fmt.Sprintf("Query server started listening on %s...", s.httpHostPort))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Start listening on http and private http port concurrently
|
||||
func (s *Server) Start(ctx context.Context) error {
|
||||
err := s.initListeners()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var httpPort int
|
||||
if port, err := utils.GetPort(s.httpConn.Addr()); err == nil {
|
||||
httpPort = port
|
||||
}
|
||||
|
||||
go func() {
|
||||
slog.Info("Starting HTTP server", "port", httpPort, "addr", s.httpHostPort)
|
||||
|
||||
switch err := s.httpServer.Serve(s.httpConn); err {
|
||||
case nil, http.ErrServerClosed, cmux.ErrListenerClosed:
|
||||
// normal exit, nothing to do
|
||||
default:
|
||||
slog.Error("Could not start HTTP server", errors.Attr(err))
|
||||
}
|
||||
s.unavailableChannel <- healthcheck.Unavailable
|
||||
}()
|
||||
|
||||
go func() {
|
||||
slog.Info("Starting OpAmp Websocket server", "addr", constants.OpAmpWsEndpoint)
|
||||
err := s.opampServer.Start(constants.OpAmpWsEndpoint)
|
||||
if err != nil {
|
||||
slog.Error("opamp ws server failed to start", errors.Attr(err))
|
||||
s.unavailableChannel <- healthcheck.Unavailable
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -125,6 +266,12 @@ func (s *Server) Start(ctx context.Context) error {
|
||||
}
|
||||
|
||||
func (s *Server) Stop(ctx context.Context) error {
|
||||
if s.httpServer != nil {
|
||||
if err := s.httpServer.Shutdown(context.Background()); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
s.opampServer.Stop()
|
||||
|
||||
return nil
|
||||
|
||||
@@ -10,7 +10,11 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
)
|
||||
|
||||
const OpAmpWsEndpoint = "0.0.0.0:4320" // address for opamp websocket
|
||||
const (
|
||||
HTTPHostPort = "0.0.0.0:8080" // Address to serve http (query service)
|
||||
PrivateHostPort = "0.0.0.0:8085" // Address to server internal services like alert manager
|
||||
OpAmpWsEndpoint = "0.0.0.0:4320" // address for opamp websocket
|
||||
)
|
||||
|
||||
const MaxAllowedPointsInTimeSeries = 300
|
||||
|
||||
|
||||
12
pkg/query-service/healthcheck/handler.go
Normal file
12
pkg/query-service/healthcheck/handler.go
Normal file
@@ -0,0 +1,12 @@
|
||||
package healthcheck
|
||||
|
||||
const (
|
||||
// Unavailable indicates the service is not able to handle requests
|
||||
Unavailable Status = iota
|
||||
// Ready indicates the service is ready to handle requests
|
||||
Ready
|
||||
// Broken indicates that the healthcheck itself is broken, not serving HTTP
|
||||
Broken
|
||||
)
|
||||
|
||||
type Status int
|
||||
@@ -91,10 +91,6 @@ type Reader interface {
|
||||
|
||||
GetMinAndMaxTimestampForTraceID(ctx context.Context, traceID []string) (int64, int64, error)
|
||||
|
||||
// Query Progress tracking helpers.
|
||||
ReportQueryStartForProgressTracking(queryId string) (reportQueryFinished func(), apiErr *model.ApiError)
|
||||
SubscribeToQueryProgress(queryId string) (<-chan model.QueryProgress, func(), *model.ApiError)
|
||||
|
||||
//trace
|
||||
GetTraceFields(ctx context.Context) (*model.GetFieldsResponse, *model.ApiError)
|
||||
UpdateTraceField(ctx context.Context, field *model.UpdateField) *model.ApiError
|
||||
|
||||
@@ -7,14 +7,6 @@ import (
|
||||
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
|
||||
)
|
||||
|
||||
type QueryProgress struct {
|
||||
ReadRows uint64 `json:"read_rows"`
|
||||
|
||||
ReadBytes uint64 `json:"read_bytes"`
|
||||
|
||||
ElapsedMs uint64 `json:"elapsed_ms"`
|
||||
}
|
||||
|
||||
func GetLogFieldsV3(ctx context.Context, queryRangeParams *v3.QueryRangeParamsV3, fields *GetFieldsResponse) map[string]v3.AttributeKey {
|
||||
data := map[string]v3.AttributeKey{}
|
||||
for _, query := range queryRangeParams.CompositeQuery.BuilderQueries {
|
||||
|
||||
@@ -10,15 +10,12 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/alertmanager"
|
||||
"github.com/SigNoz/signoz/pkg/apiserver"
|
||||
"github.com/SigNoz/signoz/pkg/apiserver/signozapiserver"
|
||||
"github.com/SigNoz/signoz/pkg/auditor"
|
||||
"github.com/SigNoz/signoz/pkg/authz"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
"github.com/SigNoz/signoz/pkg/flagger"
|
||||
"github.com/SigNoz/signoz/pkg/gateway"
|
||||
"github.com/SigNoz/signoz/pkg/global"
|
||||
"github.com/SigNoz/signoz/pkg/http/handler"
|
||||
"github.com/SigNoz/signoz/pkg/identn"
|
||||
"github.com/SigNoz/signoz/pkg/instrumentation"
|
||||
"github.com/SigNoz/signoz/pkg/licensing"
|
||||
"github.com/SigNoz/signoz/pkg/modules/aiobservability"
|
||||
@@ -45,11 +42,9 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
"github.com/SigNoz/signoz/pkg/querier"
|
||||
"github.com/SigNoz/signoz/pkg/ruler"
|
||||
"github.com/SigNoz/signoz/pkg/sharder"
|
||||
"github.com/SigNoz/signoz/pkg/statsreporter"
|
||||
"github.com/SigNoz/signoz/pkg/subscription"
|
||||
"github.com/SigNoz/signoz/pkg/types/authtypes"
|
||||
"github.com/SigNoz/signoz/pkg/web"
|
||||
"github.com/SigNoz/signoz/pkg/zeus"
|
||||
"github.com/swaggest/jsonschema-go"
|
||||
"github.com/swaggest/openapi-go"
|
||||
@@ -66,13 +61,6 @@ type OpenAPI struct {
|
||||
}
|
||||
|
||||
func NewOpenAPI(ctx context.Context, instrumentation instrumentation.Instrumentation) (*OpenAPI, error) {
|
||||
// Build the apiserver with the default config. Only route registration is
|
||||
// needed for spec generation, the server is never started.
|
||||
apiserverConfig, ok := apiserver.NewConfigFactory().New().(*apiserver.Config)
|
||||
if !ok {
|
||||
return nil, errors.NewInternalf(errors.CodeInternal, "cannot build apiserver config for openapi generation")
|
||||
}
|
||||
|
||||
apiserver, err := signozapiserver.NewFactory(
|
||||
struct{ organization.Getter }{},
|
||||
struct{ authz.AuthZ }{},
|
||||
@@ -113,14 +101,9 @@ func NewOpenAPI(ctx context.Context, instrumentation instrumentation.Instrumenta
|
||||
struct{ ruler.Handler }{},
|
||||
struct{ statsreporter.Handler }{},
|
||||
struct{ savedview.Handler }{},
|
||||
global.Config{},
|
||||
struct{ identn.IdentNResolver }{},
|
||||
struct{ sharder.Sharder }{},
|
||||
struct{ auditor.Auditor }{},
|
||||
struct{ web.Web }{},
|
||||
struct{ quickfilter.Module }{},
|
||||
struct{ quickfilter.Handler }{},
|
||||
).New(ctx, instrumentation.ToProviderSettings(), *apiserverConfig)
|
||||
).New(ctx, instrumentation.ToProviderSettings(), apiserver.Config{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -318,7 +318,7 @@ func NewQuerierProviderFactories(telemetryStore telemetrystore.TelemetryStore, p
|
||||
)
|
||||
}
|
||||
|
||||
func NewAPIServerProviderFactories(orgGetter organization.Getter, authz authz.AuthZ, modules Modules, handlers Handlers, globalConfig global.Config, gatewayService gateway.Gateway, identNResolver identn.IdentNResolver, sharder sharder.Sharder, auditor auditor.Auditor, web web.Web) factory.NamedMap[factory.ProviderFactory[apiserver.APIServer, apiserver.Config]] {
|
||||
func NewAPIServerProviderFactories(orgGetter organization.Getter, authz authz.AuthZ, modules Modules, handlers Handlers, globalConfig global.Config, gatewayService gateway.Gateway) factory.NamedMap[factory.ProviderFactory[apiserver.APIServer, apiserver.Config]] {
|
||||
return factory.MustNewNamedMap(
|
||||
signozapiserver.NewFactory(
|
||||
orgGetter,
|
||||
@@ -360,11 +360,6 @@ func NewAPIServerProviderFactories(orgGetter organization.Getter, authz authz.Au
|
||||
handlers.RulerHandler,
|
||||
handlers.StatsHandler,
|
||||
handlers.SavedView,
|
||||
globalConfig,
|
||||
identNResolver,
|
||||
sharder,
|
||||
auditor,
|
||||
web,
|
||||
modules.QuickFilter,
|
||||
handlers.QuickFilter,
|
||||
),
|
||||
|
||||
@@ -102,10 +102,6 @@ func TestNewProviderFactories(t *testing.T) {
|
||||
Handlers{},
|
||||
global.Config{},
|
||||
nil,
|
||||
nil,
|
||||
nil,
|
||||
nil,
|
||||
nil,
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -635,20 +635,13 @@ func New(
|
||||
ctx,
|
||||
providerSettings,
|
||||
config.APIServer,
|
||||
NewAPIServerProviderFactories(orgGetter, authz, modules, handlers, config.Global, gateway, identNResolver, sharder, auditor, web),
|
||||
NewAPIServerProviderFactories(orgGetter, authz, modules, handlers, config.Global, gateway),
|
||||
"signoz",
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Register the API server with the registry so its lifecycle is managed
|
||||
// alongside the other services and it shows up in the health endpoint.
|
||||
err = registry.Add(ctx, factory.NewNamedService(factory.MustNewName("apiserver"), apiserverInstance))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &SigNoz{
|
||||
Registry: registry,
|
||||
Analytics: analytics,
|
||||
|
||||
Reference in New Issue
Block a user