最後のメトリクスを取得するAPIを実装する

README.md

@@ -48,6 +48,13 @@ ErrorLogFile: "/var/log/logreport.log"
 # ログバッファサイズ
 LogBufferSize: 4096
 
+# 内部API
+API:
+  # 内部API 有効/無効
+  Enabled: false
+  # 内部APIを有効にする場合にListenするUnix Domain Socket
+  # SocketPath: "/var/run/logreport-api.sock"
+
 # 読み込むログファイルのパス
 LogFile: "/var/log/nginx/access.log"
 
@@ -167,6 +174,37 @@ Metrics:
     LogColumn: "upstream_request_time"
 ```
 
+### 内部API
+
+設定ファイルで以下のように設定することで有効にすることができます。
+```yaml
+API:
+  Enabled: true
+  SocketPath: "/var/run/logreport-api.sock"
+```
+
+#### 最新のメトリックのキー一覧
+```
+curl --unix-socket /var/run/logreport-api.sock http://localhost/keys
+```
+
+```
+["bytes_sent","hit-count.HIT","hit-count.MISS","http.request","http.status.205","http.status.300","http.status.301","http.status.410","http.status.414","http.status.426","http.status.431","http.status.504","https.request","https.status.429","upstream_request_time_max"]
+```
+
+#### 最新のメトリックを取得
+
+keysにキーを指定することで取得可能です。
+
+```
+curl --unix-socket /var/run/logreport-api.sock http://localhost/metrics?keys=http.request,https.request,http.status.200,https.status.200
+
+```
+
+```json
+{"http.request":5,"http.status.200":0,"https.request":5,"https.status.200":1,"time":"2026-03-16T21:29:42+09:00"}
+```
+
 ## ライセンス
 
 MIT

cmd/logreport/config.sample.yaml

@@ -4,6 +4,10 @@ PidFile: "/var/run/logreport.pid"
 ErrorLogFile: "/var/log/logreport.log"
 LogBufferSize: 4096
 
+API:
+  Enabled: false
+  # SocketPath: "/var/run/logreport-api.sock"
+
 # Read log file path. Posisiton file path.
 LogFile: "/var/log/nginx/access.log"
 PosFile: "/var/log/nginx/access.log.pos"

cmd/logreport/main.go

@@ -6,6 +6,8 @@ import (
 	"crypto/x509"
 	"errors"
 	"flag"
+	"net"
+	"net/http"
 	"os"
 	"os/signal"
 	"sort"
@@ -19,6 +21,7 @@ import (
 	"github.com/masa23/gotail"
 
 	"github.com/masa23/logreport"
+	"github.com/masa23/logreport/internal/api"
 	"github.com/masa23/logreport/internal/exporter"
 	"github.com/masa23/logreport/internal/exporter/graphite"
 	"github.com/masa23/logreport/internal/exporter/otlpgrpc"
@@ -30,6 +33,7 @@ var (
 	confLock         = new(sync.Mutex)
 	graphiteExporter *graphite.GraphiteExporter
 	otlpGrpcExporter *otlpgrpc.OtlpGrpcExporter
+	metricsAPI       *api.API
 )
 
 type ItemCount map[string]int64
@@ -80,6 +84,43 @@ func main() {
 	pid := os.Getpid()
 	ltsvlog.Logger.Info().Fmt("msg", "start logreport pid=%d", pid).Log()
 
+	if conf.API.Enabled {
+		f, err := os.Lstat(conf.API.SocketPath)
+		if err != nil {
+			if !os.IsNotExist(err) {
+				ltsvlog.Logger.Err(err)
+				os.Exit(1)
+			}
+		} else {
+			if f.Mode()&os.ModeSocket == 0 {
+				ltsvlog.Logger.Err(errstack.New("API socket path exists and is not a unix socket"))
+				os.Exit(1)
+			}
+			if err := os.Remove(conf.API.SocketPath); err != nil {
+				ltsvlog.Logger.Err(err)
+				os.Exit(1)
+			}
+		}
+		l, err := net.Listen("unix", conf.API.SocketPath)
+		if err != nil {
+			ltsvlog.Logger.Err(err)
+			os.Exit(1)
+		}
+		defer l.Close()
+		if err := os.Chmod(conf.API.SocketPath, 0o700); err != nil {
+			ltsvlog.Logger.Err(err)
+			os.Exit(1)
+		}
+		metricsAPI = api.NewAPI()
+		go func() {
+			if err := metricsAPI.Serve(l); err != nil {
+				if errors.Is(err, http.ErrServerClosed) {
+					return
+				}
+				ltsvlog.Logger.Err(err)
+			}
+		}()
+	}
 	if conf.Exporters.Graphite != nil || conf.Graphite != nil {
 		var exporterConfig *graphite.GraphiteExporterConfig
 		if conf.Exporters.Graphite != nil {
@@ -309,6 +350,10 @@ func readLog() {
 					}
 				}
 			}
+
+			if metricsAPI != nil {
+				go metricsAPI.SetLastMetrics(metrics)
+			}
 			lock.Unlock()
 		}
 	}()

config.go

@@ -3,6 +3,7 @@ package logreport
 import (
 	"fmt"
 	"os"
+	"strings"
 	"time"
 
 	"gopkg.in/yaml.v2"
@@ -36,6 +37,12 @@ type Config struct {
 	TimeParse     string          `yaml:"TimeParse"`
 	LogColumns    []logColumn
 	Exporters     configExporters `yaml:"Exporters"`
+	API           configAPI       `yaml:"API"`
+}
+
+type configAPI struct {
+	Enabled    bool   `yaml:"Enabled"`
+	SocketPath string `yaml:"SocketPath"`
 }
 
 type configExporters struct {
@@ -114,6 +121,13 @@ func ConfigLoad(file string) (*Config, error) {
 		return conf, fmt.Errorf("LogFormat type %s is unsupported", conf.LogFormat)
 	}
 
+	if conf.API.Enabled {
+		socketPath := strings.TrimSpace(conf.API.SocketPath)
+		if socketPath == "" {
+			return conf, fmt.Errorf("API.Enabled is true but API.SocketPath is empty")
+		}
+	}
+
 	for i, metric := range conf.Metrics {
 		if err := validateMetricAndSetDefault(&conf.Metrics[i]); err != nil {
 			return conf, err

go.mod

@@ -15,6 +15,7 @@ require (
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.34.0
 	go.opentelemetry.io/otel/sdk v1.34.0
 	go.opentelemetry.io/otel/sdk/metric v1.34.0
+	golang.org/x/sync v0.12.0
 	golang.org/x/time v0.10.0
 	google.golang.org/grpc v1.70.0
 	gopkg.in/yaml.v2 v2.4.0

go.sum

@@ -55,6 +55,8 @@ go.opentelemetry.io/proto/otlp v1.5.0 h1:xJvq7gMzB31/d406fB8U5CBdyQGw4P399D1aQWU
 go.opentelemetry.io/proto/otlp v1.5.0/go.mod h1:keN8WnHxOy8PG0rQZjJJ5A2ebUoafqWp0eVQ4yIXvJ4=
 golang.org/x/net v0.38.0 h1:vRMAPTMaeGqVhG5QyLJHqNDwecKTomGeqbnfZyKlBI8=
 golang.org/x/net v0.38.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
+golang.org/x/sync v0.12.0 h1:MHc5BpPuC30uJk597Ri8TV3CNZcTLu6B6z4lJy+g6Jw=
+golang.org/x/sync v0.12.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
 golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik=
 golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=

internal/api/api.go

@@ -0,0 +1,158 @@
+package api
+
+import (
+	"encoding/json"
+	"log"
+	"net"
+	"net/http"
+	"slices"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/hnakamur/ltsvlog"
+	"github.com/masa23/logreport/internal/exporter"
+	"golang.org/x/sync/singleflight"
+)
+
+type API struct {
+	lastMetrics   map[string]*lastMetric
+	lastTimestamp time.Time
+	mu            sync.RWMutex
+	version       int
+	sf            singleflight.Group
+}
+
+type lastMetric struct {
+	Version int
+	Metric  *exporter.Metric
+}
+
+func NewAPI() *API {
+	return &API{
+		lastMetrics: map[string]*lastMetric{},
+		mu:          sync.RWMutex{},
+		version:     0,
+		sf:          singleflight.Group{},
+	}
+}
+
+func (a *API) SetLastMetrics(metrics []*exporter.Metric) {
+	_, _, _ = a.sf.Do("setLastMetrics", func() (any, error) {
+		a.setLastMetrics(metrics)
+		return nil, nil
+	})
+}
+
+func (a *API) setLastMetrics(metrics []*exporter.Metric) {
+	lastTimestamp := time.Time{}
+	for _, m := range metrics {
+		if m.Timestamp.Compare(lastTimestamp) > 0 {
+			lastTimestamp = m.Timestamp
+		}
+	}
+
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	a.lastTimestamp = lastTimestamp
+	a.version++
+	for _, m := range metrics {
+		// NOTE: metricsには複数のtimestampのメトリクスが入る可能性があるため、
+		// その中で最新の時刻のメトリクスを採用する
+		if !m.Timestamp.Equal(lastTimestamp) {
+			continue
+		}
+		key := m.ItemName
+		if m.ItemValue != "" {
+			key += "." + m.ItemValue
+		}
+		lm, ok := a.lastMetrics[key]
+		if !ok {
+			lm = &lastMetric{}
+		}
+
+		lm.Version = a.version
+		lm.Metric = m
+		a.lastMetrics[key] = lm
+	}
+	for k, v := range a.lastMetrics {
+		if v.Version != a.version {
+			delete(a.lastMetrics, k)
+		}
+	}
+}
+
+func (a *API) Serve(l net.Listener) error {
+	mu := http.NewServeMux()
+	mu.HandleFunc("/keys", a.getKeys)
+	mu.HandleFunc("/metrics", a.getLastMetrics)
+	sv := http.Server{
+		Handler:           mu,
+		ReadHeaderTimeout: 5 * time.Second,
+		ReadTimeout:       10 * time.Second,
+		WriteTimeout:      10 * time.Second,
+		IdleTimeout:       60 * time.Second,
+	}
+	return sv.Serve(l)
+}
+
+func (a *API) recover() {
+	if v := recover(); v != nil {
+		log.Printf("PANIC: %v", v)
+		return
+	}
+}
+
+func (a *API) getKeys(w http.ResponseWriter, r *http.Request) {
+	defer a.recover()
+
+	keys := make([]string, 0, len(a.lastMetrics))
+	a.mu.RLock()
+	for k := range a.lastMetrics {
+		keys = append(keys, k)
+	}
+	a.mu.RUnlock()
+
+	slices.Sort(keys)
+	w.WriteHeader(http.StatusOK)
+	if err := json.NewEncoder(w).Encode(&keys); err != nil {
+		ltsvlog.Logger.Err(err)
+	}
+}
+
+func (a *API) getLastMetrics(w http.ResponseWriter, r *http.Request) {
+	defer a.recover()
+
+	keysStr := r.URL.Query().Get("keys")
+	var keys []string
+	if keysStr == "" {
+		keys = []string{}
+	} else {
+		keys = strings.Split(keysStr, ",")
+	}
+
+	res := map[string]any{}
+	a.mu.RLock()
+	for _, k := range keys {
+		k = strings.TrimSpace(k)
+		if k == "" {
+			continue
+		}
+
+		var v any = 0
+		m, ok := a.lastMetrics[k]
+		if ok && m.Metric.Value != nil {
+			v = m.Metric.Value
+		}
+		res[k] = v
+	}
+	if !a.lastTimestamp.IsZero() {
+		res["time"] = a.lastTimestamp
+	}
+	a.mu.RUnlock()
+
+	w.WriteHeader(http.StatusOK)
+	if err := json.NewEncoder(w).Encode(&res); err != nil {
+		ltsvlog.Logger.Err(err)
+	}
+}