shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: expose daily metric query api

Browse Source
This commit is contained in:
lingniu
2026-07-01 23:30:54 +08:00
parent c2d70dad86
commit 1a53011408
4 changed files with 373 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
deploy/portainer/docker-compose-go.yml
View File

@@ -95,6 +95,7 @@ services:
REDIS_PASSWORD: ${REDIS_PASSWORD:?set REDIS_PASSWORD}
REDIS_DB: ${REDIS_DB:-50}
ONLINE_TTL_SECONDS: ${ONLINE_TTL_SECONDS:-600}
MYSQL_DSN: ${MYSQL_DSN:-}
ports:
- "${GO_REALTIME_HTTP_PORT:-20210}:20210"

32
go/vehicle-gateway/cmd/realtime-api/main.go
View File

@@ -2,6 +2,7 @@ package main
import (
"context"
"database/sql"
"encoding/json"
"net/http"
"os"
@@ -11,12 +12,14 @@ import (
"syscall"
"time"
_ "github.com/go-sql-driver/mysql"
"github.com/redis/go-redis/v9"
"github.com/segmentio/kafka-go"
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime"
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats"
)
func main() {
@@ -45,9 +48,36 @@ func main() {
logger.Warn("KAFKA_BROKERS is empty; realtime api will serve existing redis data only")
}
mux := http.NewServeMux()
mux.Handle("/api/realtime/vehicles/", realtime.NewHandler(repository))
closeStats := func() {}
if dsn := strings.TrimSpace(os.Getenv("MYSQL_DSN")); dsn != "" {
db, err := sql.Open("mysql", dsn)
if err != nil {
logger.Error("stats mysql open failed", "error", err)
os.Exit(1)
}
if err := db.PingContext(ctx); err != nil {
_ = db.Close()
logger.Error("stats mysql ping failed", "error", err)
os.Exit(1)
}
closeStats = func() { _ = db.Close() }
mux.Handle("/api/stats/daily-metrics", stats.NewMetricHandler(stats.NewMetricRepository(db)))
logger.Info("stats mysql query enabled")
} else {
mux.HandleFunc("/api/stats/daily-metrics", func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusServiceUnavailable)
_ = json.NewEncoder(w).Encode(map[string]any{"error": "MYSQL_DSN is not configured"})
})
logger.Warn("MYSQL_DSN is empty; stats query api disabled")
}
defer closeStats()
server := &http.Server{
Addr: env("HTTP_ADDR", ":20210"),
Handler: realtime.NewHandler(repository),
Handler: mux,
ReadHeaderTimeout: 5 * time.Second,
}
go func() {

241
go/vehicle-gateway/internal/stats/query.go Normal file
View File

@@ -0,0 +1,241 @@
package stats
import (
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"net/http"
"strconv"
"strings"
)
type Queryer interface {
QueryContext(context.Context, string, ...any) (*sql.Rows, error)
}
type MetricQuery struct {
VIN string
Protocol string
MetricKey string
DateFrom string
DateTo string
Limit int
Offset int
}
type MetricRow struct {
VIN string `json:"vin"`
StatDate string `json:"stat_date"`
Protocol string `json:"protocol"`
MetricKey string `json:"metric_key"`
MetricValue float64 `json:"metric_value"`
MetricUnit string `json:"metric_unit"`
FirstTotalMileageKM *float64 `json:"first_total_mileage_km,omitempty"`
LatestTotalMileageKM *float64 `json:"latest_total_mileage_km,omitempty"`
SampleCount int64 `json:"sample_count"`
CalculationMethod string `json:"calculation_method"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
}
type MetricRepository struct {
db Queryer
}
func NewMetricRepository(db Queryer) *MetricRepository {
if db == nil {
panic("metric query db must not be nil")
}
return &MetricRepository{db: db}
}
func (r *MetricRepository) Query(ctx context.Context, query MetricQuery) ([]MetricRow, error) {
query = normalizeMetricQuery(query)
sqlText, args := buildMetricSQL(query)
rows, err := r.db.QueryContext(ctx, sqlText, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var out []MetricRow
for rows.Next() {
var row MetricRow
var first sql.NullFloat64
var latest sql.NullFloat64
if err := rows.Scan(
&row.VIN,
&row.StatDate,
&row.Protocol,
&row.MetricKey,
&row.MetricValue,
&row.MetricUnit,
&first,
&latest,
&row.SampleCount,
&row.CalculationMethod,
&row.CreatedAt,
&row.UpdatedAt,
); err != nil {
return nil, err
}
if first.Valid {
row.FirstTotalMileageKM = &first.Float64
}
if latest.Valid {
row.LatestTotalMileageKM = &latest.Float64
}
out = append(out, row)
}
return out, rows.Err()
}
func normalizeMetricQuery(query MetricQuery) MetricQuery {
query.VIN = strings.TrimSpace(query.VIN)
query.Protocol = strings.ToUpper(strings.TrimSpace(query.Protocol))
query.MetricKey = strings.TrimSpace(query.MetricKey)
query.DateFrom = strings.TrimSpace(query.DateFrom)
query.DateTo = strings.TrimSpace(query.DateTo)
if query.Limit <= 0 {
query.Limit = 50
}
return query
}
func buildMetricSQL(query MetricQuery) (string, []any) {
var where []string
var args []any
add := func(clause string, value any) {
where = append(where, clause)
args = append(args, value)
}
if query.VIN != "" {
add("vin = ?", query.VIN)
}
if query.Protocol != "" {
add("protocol = ?", query.Protocol)
}
if query.MetricKey != "" {
add("metric_key = ?", query.MetricKey)
}
if query.DateFrom != "" {
add("stat_date >= ?", query.DateFrom)
}
if query.DateTo != "" {
add("stat_date <= ?", query.DateTo)
}
sqlText := `SELECT vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric`
if len(where) > 0 {
sqlText += " WHERE " + strings.Join(where, " AND ")
}
sqlText += " ORDER BY stat_date DESC, vin ASC, protocol ASC, metric_key ASC LIMIT ? OFFSET ?"
args = append(args, query.Limit, query.Offset)
return sqlText, args
}
type MetricHandler struct {
repository *MetricRepository
}
func NewMetricHandler(repository *MetricRepository) *MetricHandler {
if repository == nil {
panic("metric repository must not be nil")
}
return &MetricHandler{repository: repository}
}
func (h *MetricHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
writeMetricError(w, http.StatusMethodNotAllowed, "method not allowed")
return
}
if strings.Trim(r.URL.Path, "/") != "api/stats/daily-metrics" {
writeMetricError(w, http.StatusNotFound, "route not found")
return
}
query, err := parseMetricQuery(r)
if err != nil {
writeMetricError(w, http.StatusBadRequest, err.Error())
return
}
rows, err := h.repository.Query(r.Context(), query)
if err != nil {
writeMetricError(w, http.StatusInternalServerError, err.Error())
return
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{
"items": rows,
"total": len(rows),
"limit": query.Limit,
"offset": query.Offset,
})
}
func parseMetricQuery(r *http.Request) (MetricQuery, error) {
values := r.URL.Query()
limit, err := parseBoundedInt(values.Get("limit"), 50, 1, 1000, "limit")
if err != nil {
return MetricQuery{}, err
}
offset, err := parseBoundedInt(values.Get("offset"), 0, 0, 1_000_000, "offset")
if err != nil {
return MetricQuery{}, err
}
query := MetricQuery{
VIN: values.Get("vin"),
Protocol: values.Get("protocol"),
MetricKey: values.Get("metricKey"),
DateFrom: values.Get("dateFrom"),
DateTo: values.Get("dateTo"),
Limit: limit,
Offset: offset,
}
if !validDate(query.DateFrom) || !validDate(query.DateTo) {
return MetricQuery{}, errors.New("dateFrom/dateTo must use YYYY-MM-DD")
}
return normalizeMetricQuery(query), nil
}
func parseBoundedInt(raw string, fallback int, min int, max int, name string) (int, error) {
raw = strings.TrimSpace(raw)
if raw == "" {
return fallback, nil
}
value, err := strconv.Atoi(raw)
if err != nil || value < min || value > max {
return 0, fmt.Errorf("%s must be between %d and %d", name, min, max)
}
return value, nil
}
func validDate(value string) bool {
value = strings.TrimSpace(value)
if value == "" {
return true
}
if len(value) != len("2006-01-02") {
return false
}
for i, r := range value {
switch i {
case 4, 7:
if r != '-' {
return false
}
default:
if r < '0' || r > '9' {
return false
}
}
}
return true
}
func writeMetricError(w http.ResponseWriter, status int, message string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(map[string]any{"error": message})
}

100
go/vehicle-gateway/internal/stats/query_test.go Normal file
View File

@@ -0,0 +1,100 @@
package stats
import (
"context"
"database/sql"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/DATA-DOG/go-sqlmock"
)
func TestMetricRepositoryQueriesDailyMetricsWithFilters(t *testing.T) {
db, mock, err := sqlmock.New()
if err != nil {
t.Fatalf("sqlmock.New() error = %v", err)
}
defer db.Close()
mock.ExpectQuery("SELECT vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric").
WithArgs("LKLG7C4E3NA774736", "JT808", "2026-07-01", "2026-07-01", 20, 0).
WillReturnRows(sqlmock.NewRows([]string{
"vin", "stat_date", "protocol", "metric_key", "metric_value", "metric_unit",
"first_total_mileage_km", "latest_total_mileage_km", "sample_count", "calculation_method",
"created_at", "updated_at",
}).AddRow(
"LKLG7C4E3NA774736", "2026-07-01", "JT808", MetricDailyTotalMileageKM, 12345.6, "km",
12345.6, 12345.6, 13, "TOTAL_MILEAGE_DIFF", "2026-07-01 22:49:11", "2026-07-01 23:09:36",
))
repository := NewMetricRepository(db)
rows, err := repository.Query(context.Background(), MetricQuery{
VIN: "LKLG7C4E3NA774736",
Protocol: "JT808",
DateFrom: "2026-07-01",
DateTo: "2026-07-01",
Limit: 20,
})
if err != nil {
t.Fatalf("Query() error = %v", err)
}
if len(rows) != 1 {
t.Fatalf("row count = %d", len(rows))
}
if rows[0].MetricKey != MetricDailyTotalMileageKM || rows[0].MetricValue != 12345.6 {
t.Fatalf("unexpected row: %#v", rows[0])
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatalf("sql expectations: %v", err)
}
}
func TestMetricHandlerReturnsDailyMetrics(t *testing.T) {
db, mock, err := sqlmock.New()
if err != nil {
t.Fatalf("sqlmock.New() error = %v", err)
}
defer db.Close()
mock.ExpectQuery("SELECT vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric").
WithArgs("LB9A32A21R0LS1707", "GB32960", "2020-07-01", "2020-07-01", 50, 0).
WillReturnRows(sqlmock.NewRows([]string{
"vin", "stat_date", "protocol", "metric_key", "metric_value", "metric_unit",
"first_total_mileage_km", "latest_total_mileage_km", "sample_count", "calculation_method",
"created_at", "updated_at",
}).AddRow(
"LB9A32A21R0LS1707", "2020-07-01", "GB32960", MetricDailyMileageKM, 0.0, "km",
53490.9, 53490.9, 3, "TOTAL_MILEAGE_DIFF", "2026-07-01 22:07:58", "2026-07-01 22:28:25",
))
handler := NewMetricHandler(NewMetricRepository(db))
request := httptest.NewRequest(http.MethodGet, "/api/stats/daily-metrics?vin=LB9A32A21R0LS1707&protocol=GB32960&dateFrom=2020-07-01&dateTo=2020-07-01", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK {
t.Fatalf("status = %d body=%s", response.Code, response.Body.String())
}
body := response.Body.String()
for _, want := range []string{`"vin":"LB9A32A21R0LS1707"`, `"metric_key":"daily_mileage_km"`, `"total":1`} {
if !strings.Contains(body, want) {
t.Fatalf("response missing %s: %s", want, body)
}
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatalf("sql expectations: %v", err)
}
}
func TestMetricHandlerRejectsInvalidPagination(t *testing.T) {
handler := NewMetricHandler(NewMetricRepository(&sql.DB{}))
request := httptest.NewRequest(http.MethodGet, "/api/stats/daily-metrics?limit=2001", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusBadRequest {
t.Fatalf("status = %d body=%s", response.Code, response.Body.String())
}
}