wip
This commit is contained in:
66
server/clickhouse.go
Normal file
66
server/clickhouse.go
Normal file
@@ -0,0 +1,66 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2"
|
||||
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
|
||||
"go.ntppool.org/data-api/version"
|
||||
"golang.org/x/exp/slog"
|
||||
)
|
||||
|
||||
func (srv *Server) chConn(ctx context.Context) (driver.Conn, error) {
|
||||
|
||||
conn, err := clickhouse.Open(&clickhouse.Options{
|
||||
Addr: []string{"10.43.207.123:9000"},
|
||||
Auth: clickhouse.Auth{
|
||||
Database: "geodns3",
|
||||
Username: "default",
|
||||
Password: "",
|
||||
},
|
||||
// Debug: true,
|
||||
// Debugf: func(format string, v ...interface{}) {
|
||||
// slog.Info("debug format", "format", format)
|
||||
// fmt.Printf(format+"\n", v)
|
||||
// },
|
||||
Settings: clickhouse.Settings{
|
||||
"max_execution_time": 60,
|
||||
},
|
||||
Compression: &clickhouse.Compression{
|
||||
Method: clickhouse.CompressionLZ4,
|
||||
},
|
||||
DialTimeout: time.Second * 5,
|
||||
MaxOpenConns: 5,
|
||||
MaxIdleConns: 5,
|
||||
ConnMaxLifetime: time.Duration(10) * time.Minute,
|
||||
ConnOpenStrategy: clickhouse.ConnOpenInOrder,
|
||||
BlockBufferSize: 10,
|
||||
MaxCompressionBuffer: 10240,
|
||||
ClientInfo: clickhouse.ClientInfo{
|
||||
Products: []struct {
|
||||
Name string
|
||||
Version string
|
||||
}{
|
||||
{Name: "data-api", Version: version.Version()},
|
||||
},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
v, err := conn.ServerVersion()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
slog.Info("clickhouse connection", "version", v)
|
||||
|
||||
err = conn.Ping(ctx)
|
||||
if err != nil {
|
||||
slog.Error("clickhouse ping", "err", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return conn, nil
|
||||
}
|
||||
144
server/server.go
Normal file
144
server/server.go
Normal file
@@ -0,0 +1,144 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
|
||||
chdb "go.ntppool.org/data-api/chdb"
|
||||
"go.ntppool.org/data-api/ntpdb"
|
||||
|
||||
"github.com/labstack/echo/v4"
|
||||
"github.com/labstack/echo/v4/middleware"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||
"go.opentelemetry.io/contrib/instrumentation/github.com/labstack/echo/otelecho"
|
||||
otrace "go.opentelemetry.io/otel/trace"
|
||||
"golang.org/x/exp/slog"
|
||||
"golang.org/x/sync/errgroup"
|
||||
)
|
||||
|
||||
type Server struct {
|
||||
db *sql.DB
|
||||
ch *chdb.ClickHouse
|
||||
|
||||
ctx context.Context
|
||||
mr *prometheus.Registry
|
||||
tracer otrace.Tracer
|
||||
}
|
||||
|
||||
func NewServer(ctx context.Context) (*Server, error) {
|
||||
mr := prometheus.NewRegistry()
|
||||
|
||||
ch, err := chdb.New("database.yaml")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("clickhouse open: %w", err)
|
||||
}
|
||||
db, err := ntpdb.OpenDB("database.yaml")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("mysql open: %w", err)
|
||||
}
|
||||
|
||||
srv := &Server{
|
||||
ch: ch,
|
||||
db: db,
|
||||
ctx: ctx,
|
||||
mr: mr,
|
||||
}
|
||||
|
||||
err = srv.initTracer()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
srv.tracer = srv.NewTracer()
|
||||
return srv, nil
|
||||
}
|
||||
|
||||
func (srv *Server) metricsHandler() http.Handler {
|
||||
return promhttp.HandlerFor(srv.mr, promhttp.HandlerOpts{
|
||||
ErrorLog: log.Default(),
|
||||
Registry: srv.mr,
|
||||
EnableOpenMetrics: true,
|
||||
})
|
||||
}
|
||||
|
||||
func (srv *Server) Run() error {
|
||||
slog.Info("Run()")
|
||||
|
||||
ctx, cancel := context.WithCancel(srv.ctx)
|
||||
defer cancel()
|
||||
|
||||
g, _ := errgroup.WithContext(ctx)
|
||||
|
||||
metricsServer := &http.Server{
|
||||
Addr: ":9000",
|
||||
Handler: srv.metricsHandler(),
|
||||
}
|
||||
|
||||
g.Go(func() error {
|
||||
err := metricsServer.ListenAndServe()
|
||||
if err != nil {
|
||||
return fmt.Errorf("metrics server: %w", err)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
e := echo.New()
|
||||
e.Use(otelecho.Middleware("data-api"))
|
||||
|
||||
e.Use(middleware.CORSWithConfig(middleware.CORSConfig{
|
||||
AllowOrigins: []string{"http://localhost", "http://localhost:5173", "https://www.ntppool.org"},
|
||||
AllowHeaders: []string{echo.HeaderOrigin, echo.HeaderContentType, echo.HeaderAccept},
|
||||
}))
|
||||
|
||||
e.GET("/hello", func(c echo.Context) error {
|
||||
return c.String(http.StatusOK, "Hello")
|
||||
})
|
||||
|
||||
e.GET("/api/usercc", srv.userCountryData)
|
||||
|
||||
g.Go(func() error {
|
||||
return e.Start(":8000")
|
||||
})
|
||||
|
||||
return g.Wait()
|
||||
}
|
||||
|
||||
func (srv *Server) userCountryData(c echo.Context) error {
|
||||
|
||||
ctx := c.Request().Context()
|
||||
|
||||
conn, err := srv.chConn(ctx)
|
||||
if err != nil {
|
||||
slog.Error("could not connect to clickhouse", "err", err)
|
||||
return c.String(http.StatusInternalServerError, "clickhouse error")
|
||||
}
|
||||
|
||||
q := ntpdb.New(srv.db)
|
||||
zoneStats, err := q.GetZoneStats(ctx)
|
||||
if err != nil {
|
||||
slog.Error("GetZoneStats", "err", err)
|
||||
return c.String(http.StatusInternalServerError, err.Error())
|
||||
}
|
||||
if zoneStats == nil {
|
||||
slog.Info("didn't get zoneStats")
|
||||
}
|
||||
|
||||
data, err := srv.ch.UserCountryData(c.Request().Context(), conn)
|
||||
if err != nil {
|
||||
slog.Error("UserCountryData", "err", err)
|
||||
return c.String(http.StatusInternalServerError, err.Error())
|
||||
}
|
||||
|
||||
return c.JSON(http.StatusOK, struct {
|
||||
UserCountry *chdb.UserCountry
|
||||
ZoneStats *ntpdb.ZoneStats
|
||||
}{
|
||||
UserCountry: data,
|
||||
ZoneStats: zoneStats,
|
||||
})
|
||||
|
||||
}
|
||||
110
server/tracing.go
Normal file
110
server/tracing.go
Normal file
@@ -0,0 +1,110 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"os"
|
||||
|
||||
"go.ntppool.org/data-api/version"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/exporters/jaeger"
|
||||
"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
|
||||
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
|
||||
"go.opentelemetry.io/otel/propagation"
|
||||
"go.opentelemetry.io/otel/sdk/resource"
|
||||
sdktrace "go.opentelemetry.io/otel/sdk/trace"
|
||||
semconv "go.opentelemetry.io/otel/semconv/v1.18.0"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
)
|
||||
|
||||
// https://github.com/open-telemetry/opentelemetry-go-contrib/blob/main/instrumentation/github.com/labstack/echo/otelecho/example/server.go
|
||||
// https://github.com/open-telemetry/opentelemetry-go/blob/main/exporters/otlp/otlptrace/otlptracehttp/example_test.go
|
||||
|
||||
func (srv *Server) NewTracer() trace.Tracer {
|
||||
traceProvider := otel.GetTracerProvider()
|
||||
return traceProvider.Tracer("data-api")
|
||||
}
|
||||
|
||||
func (srv *Server) initTracer() error {
|
||||
|
||||
// exporter, err := srv.newStdoutExporter(os.Stdout)
|
||||
|
||||
var err error
|
||||
var exporter sdktrace.SpanExporter
|
||||
|
||||
if otlp := os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"); len(otlp) > 0 {
|
||||
exporter, err = srv.newOLTPExporter()
|
||||
} else if jaegerHost := os.Getenv("OTEL_EXPORTER_JAEGER_ENDPOINT"); len(jaegerHost) > 0 {
|
||||
exporter, err = srv.newJaegerExporter()
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if exporter != nil {
|
||||
tp := sdktrace.NewTracerProvider(
|
||||
sdktrace.WithSampler(sdktrace.AlwaysSample()),
|
||||
sdktrace.WithBatcher(exporter),
|
||||
sdktrace.WithResource(srv.newResource("prod")),
|
||||
)
|
||||
|
||||
otel.SetTracerProvider(tp)
|
||||
otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{}))
|
||||
} else {
|
||||
log.Println("tracing not configured")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (srv *Server) newJaegerExporter() (sdktrace.SpanExporter, error) {
|
||||
|
||||
exporter, err := jaeger.New(jaeger.WithAgentEndpoint())
|
||||
if err != nil {
|
||||
log.Fatalf("creating jaeger trace exporter: %v", err)
|
||||
}
|
||||
return exporter, err
|
||||
}
|
||||
|
||||
func (srv *Server) newOLTPExporter() (sdktrace.SpanExporter, error) {
|
||||
ctx := context.TODO()
|
||||
client := otlptracehttp.NewClient()
|
||||
exporter, err := otlptrace.New(ctx, client)
|
||||
if err != nil {
|
||||
log.Fatalf("creating OTLP trace exporter: %v", err)
|
||||
}
|
||||
return exporter, err
|
||||
}
|
||||
|
||||
// func (srv *Server) newStdoutExporter(w io.Writer) (sdktrace.SpanExporter, error) {
|
||||
// return stdouttrace.New(
|
||||
// stdouttrace.WithWriter(w),
|
||||
// // Use human-readable output.
|
||||
// stdouttrace.WithPrettyPrint(),
|
||||
// // Do not print timestamps for the demo.
|
||||
// stdouttrace.WithoutTimestamps(),
|
||||
// )
|
||||
// }
|
||||
|
||||
// newResource returns a resource describing this application.
|
||||
func (srv *Server) newResource(depEnv string) *resource.Resource {
|
||||
|
||||
r, err := resource.Merge(
|
||||
resource.Default(),
|
||||
resource.NewWithAttributes(
|
||||
semconv.SchemaURL,
|
||||
semconv.ServiceNameKey.String("data-api"),
|
||||
semconv.ServiceVersionKey.String(version.Version()),
|
||||
attribute.String("environment", depEnv),
|
||||
),
|
||||
)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
log.Printf("merged resource: %+v", r)
|
||||
|
||||
return r
|
||||
}
|
||||
Reference in New Issue
Block a user