From 606fa6591d75670ace8f5d34688d2c6d9032863a Mon Sep 17 00:00:00 2001 From: Ankit Anand Date: Sat, 22 May 2021 13:35:30 +0530 Subject: [PATCH] added test folder for testing interface --- .../app/clickhouseReader/reader.go | 76 +++++++++++++++++++ pkg/query-service/app/druidReader/reader.go | 22 ++++++ pkg/query-service/app/fatory.go | 12 +++ .../test/clickhouse/clickhouse.go | 20 +++++ pkg/query-service/test/druid/druid.go | 20 +++++ pkg/query-service/test/test.go | 27 +++++++ 6 files changed, 177 insertions(+) create mode 100644 pkg/query-service/app/clickhouseReader/reader.go create mode 100644 pkg/query-service/app/druidReader/reader.go create mode 100644 pkg/query-service/app/fatory.go create mode 100644 pkg/query-service/test/clickhouse/clickhouse.go create mode 100644 pkg/query-service/test/druid/druid.go create mode 100644 pkg/query-service/test/test.go diff --git a/pkg/query-service/app/clickhouseReader/reader.go b/pkg/query-service/app/clickhouseReader/reader.go new file mode 100644 index 0000000000..fa2dbf7fe2 --- /dev/null +++ b/pkg/query-service/app/clickhouseReader/reader.go @@ -0,0 +1,76 @@ +package clickhouseReader + +import ( + "context" + "database/sql" + "errors" + "fmt" + "time" +) + +const ( + minTimespanForProgressiveSearch = time.Hour + minTimespanForProgressiveSearchMargin = time.Minute + maxProgressiveSteps = 4 +) + +var ( + ErrNoOperationsTable = errors.New("no operations table supplied") + ErrNoIndexTable = errors.New("no index table supplied") + ErrStartTimeRequired = errors.New("start time is required for search queries") +) + +// SpanWriter for reading spans from ClickHouse +type TraceReader struct { + db *sql.DB + operationsTable string + indexTable string + spansTable string +} + +// NewTraceReader returns a TraceReader for the database +func NewTraceReader(db *sql.DB, operationsTable, indexTable, spansTable string) *TraceReader { + return &TraceReader{ + db: db, + operationsTable: operationsTable, + indexTable: indexTable, + spansTable: spansTable, + } +} + +func (r *TraceReader) getStrings(ctx context.Context, sql string, args ...interface{}) ([]string, error) { + rows, err := r.db.QueryContext(ctx, sql, args...) + if err != nil { + return nil, err + } + + defer rows.Close() + + values := []string{} + + for rows.Next() { + var value string + if err := rows.Scan(&value); err != nil { + return nil, err + } + values = append(values, value) + } + + if err := rows.Err(); err != nil { + return nil, err + } + + return values, nil +} + +// GetServices fetches the sorted service list that have not expired +func (r *TraceReader) GetServices(ctx context.Context) ([]string, error) { + + if r.operationsTable == "" { + return nil, ErrNoOperationsTable + } + + query := fmt.Sprintf("SELECT service FROM %s GROUP BY service", r.operationsTable) + + return r.getStrings(ctx, query) +} diff --git a/pkg/query-service/app/druidReader/reader.go b/pkg/query-service/app/druidReader/reader.go new file mode 100644 index 0000000000..e3b92d82f4 --- /dev/null +++ b/pkg/query-service/app/druidReader/reader.go @@ -0,0 +1,22 @@ +package druidReader + +import ( + "go.signoz.io/query-service/druidQuery" + "go.signoz.io/query-service/model" +) + +type DruidReader struct { + Client + SqlClient +} + +func NewSpanReader() { + initialize() +} +func initialize() { + +} + +func (druid *Druid) GetServices(client, query *model.GetServicesParams) { + return druidQuery.GetServices(druid.sqlClient, query) +} diff --git a/pkg/query-service/app/fatory.go b/pkg/query-service/app/fatory.go new file mode 100644 index 0000000000..42f42d2236 --- /dev/null +++ b/pkg/query-service/app/fatory.go @@ -0,0 +1,12 @@ +// Reader finds and loads traces and other data from storage. +type QueryReader interface { + // GetTrace(ctx context.Context, traceID model.TraceID) (*model.Trace, error) + GetServices(ctx context.Context) + // GetOperations(ctx context.Context, query OperationQueryParameters) ([]Operation, error) + // FindTraces(ctx context.Context, query *TraceQueryParameters) ([]*model.Trace, error) + // FindTraceIDs(ctx context.Context, query *TraceQueryParameters) ([]model.TraceID, error) +} + +func NewQueryReader() { + +} diff --git a/pkg/query-service/test/clickhouse/clickhouse.go b/pkg/query-service/test/clickhouse/clickhouse.go new file mode 100644 index 0000000000..6dff517a86 --- /dev/null +++ b/pkg/query-service/test/clickhouse/clickhouse.go @@ -0,0 +1,20 @@ +package clickhouse + +type ClickHouseReader struct { + ClickHouseClientUrl string +} + +func connect() string { + return "Connected to ClickHouse" +} + +func NewSpanReader() *ClickHouseReader { + connect() + return &ClickHouseReader{ + ClickHouseClientUrl: "http://localhost:9000", + } +} + +func (chReader *ClickHouseReader) GetServices() string { + return "Hello from ClickHouse" +} diff --git a/pkg/query-service/test/druid/druid.go b/pkg/query-service/test/druid/druid.go new file mode 100644 index 0000000000..252966944a --- /dev/null +++ b/pkg/query-service/test/druid/druid.go @@ -0,0 +1,20 @@ +package druid + +type DruidReader struct { + DruidClientUrl string +} + +func connect() string { + return "Connected to Druid" +} + +func NewSpanReader() *DruidReader { + connect() + return &DruidReader{ + DruidClientUrl: "http://localhost:8888", + } +} + +func (druidReader *DruidReader) GetServices() string { + return "Hello from Druid" +} diff --git a/pkg/query-service/test/test.go b/pkg/query-service/test/test.go new file mode 100644 index 0000000000..755851a374 --- /dev/null +++ b/pkg/query-service/test/test.go @@ -0,0 +1,27 @@ +package main + +import ( + "fmt" + "os" + + "go.signoz.io/query-service/test/clickhouse" + "go.signoz.io/query-service/test/druid" +) + +type StorageReader interface { + GetServices() string +} + +func main() { + storage := os.Getenv("STORAGE") + var client StorageReader + + if storage == "druid" { + client = druid.NewSpanReader() + } else if storage == "clickhouse" { + client = clickhouse.NewSpanReader() + } + + services := client.GetServices() + fmt.Println(services) +}