diff --git a/README.md b/README.md index f42a3da..0031ada 100644 --- a/README.md +++ b/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 diff --git a/cmd/logreport/config.sample.yaml b/cmd/logreport/config.sample.yaml index ca19d58..5612b7e 100644 --- a/cmd/logreport/config.sample.yaml +++ b/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" diff --git a/cmd/logreport/main.go b/cmd/logreport/main.go index 61094cd..75e7510 100644 --- a/cmd/logreport/main.go +++ b/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() } }() diff --git a/config.go b/config.go index 7dd8ec8..e13d7f3 100644 --- a/config.go +++ b/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 diff --git a/go.mod b/go.mod index 0bdadf7..c8f2793 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 95604cc..3138151 100644 --- a/go.sum +++ b/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= diff --git a/internal/api/api.go b/internal/api/api.go new file mode 100644 index 0000000..01c7c78 --- /dev/null +++ b/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) + } +}