add POST /logs installer log relay to VictoriaLogs
A host being PXE-discovered or installed is not in Kubernetes, so vlagent cannot collect its logs and a failed install leaves no record; the installer environment has no internal-CA trust or credentials for the HTTPS log ingest, and bootapi is already the plain-HTTP broker it can reach. - add POST /logs, token-guarded like POST /provisioned, relaying ndjson to vlinsert's jsonline endpoint keyed on serial+phase - stamp observed source IP and resolved NetBox device name into extra_fields - return 202 on a sink failure so logs never block an install - add BOOTAPI_VLINSERT_URL/_TIMEOUT and bootapi_log_relay metrics
This commit is contained in:
+113
-1
@@ -3,12 +3,16 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/subtle"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -30,9 +34,15 @@ type Server struct {
|
||||
// fallback is the unknown-MAC iPXE behavior: "local" (safe default) or
|
||||
// "shell" (debug).
|
||||
fallback string
|
||||
// provisionToken guards POST /provisioned; empty disables the endpoint.
|
||||
// provisionToken guards POST /provisioned and POST /logs; empty disables
|
||||
// both endpoints.
|
||||
provisionToken string
|
||||
|
||||
// vlinsertURL is the VictoriaLogs vlinsert base POST /logs relays to;
|
||||
// empty disables the endpoint. vlClient bounds the single forward attempt.
|
||||
vlinsertURL string
|
||||
vlClient *http.Client
|
||||
|
||||
// TLS listener (optional); the plain-HTTP listener is always on.
|
||||
tlsAddr string
|
||||
tlsCert string
|
||||
@@ -48,6 +58,8 @@ type Options struct {
|
||||
GitStats gitStats
|
||||
UnknownMACFallback string
|
||||
ProvisionToken string
|
||||
VLInsertURL string
|
||||
VLInsertTimeout time.Duration
|
||||
TLSAddr string
|
||||
TLSCertFile string
|
||||
TLSKeyFile string
|
||||
@@ -59,12 +71,18 @@ func New(o Options) *Server {
|
||||
if fb == "" {
|
||||
fb = "local"
|
||||
}
|
||||
timeout := o.VLInsertTimeout
|
||||
if timeout <= 0 {
|
||||
timeout = 5 * time.Second
|
||||
}
|
||||
return &Server{
|
||||
nb: o.NetBox,
|
||||
engine: o.Engine,
|
||||
metrics: newMetrics(o.Cache, o.GitStats),
|
||||
fallback: fb,
|
||||
provisionToken: o.ProvisionToken,
|
||||
vlinsertURL: strings.TrimRight(o.VLInsertURL, "/"),
|
||||
vlClient: &http.Client{Timeout: timeout},
|
||||
tlsAddr: o.TLSAddr,
|
||||
tlsCert: o.TLSCertFile,
|
||||
tlsKey: o.TLSKeyFile,
|
||||
@@ -92,6 +110,10 @@ func (s *Server) Router() http.Handler {
|
||||
// End-of-kickstart callback: flips pxe_enabled off in NetBox. Token-guarded.
|
||||
r.Post("/provisioned/{ident}", s.handleProvisioned)
|
||||
|
||||
// Installer log relay: a host being installed is not in Kubernetes, so
|
||||
// vlagent cannot collect its logs. Token-guarded like /provisioned.
|
||||
r.Post("/logs", s.handleLogs)
|
||||
|
||||
return r
|
||||
}
|
||||
|
||||
@@ -253,6 +275,96 @@ func (s *Server) handleProvisioned(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
// maxLogBody caps a relayed batch. An install's log stream is a handful of
|
||||
// lines per phase; bootapi is a relay, not a log buffer.
|
||||
const maxLogBody = 1 << 20 // 1 MiB
|
||||
|
||||
// handleLogs relays newline-delimited JSON log records from a host that is
|
||||
// PXE-discovering or installing into VictoriaLogs. Such a host is not in
|
||||
// Kubernetes (no vlagent) and has no internal-CA trust or credentials for the
|
||||
// HTTPS log ingest, so bootapi — the plain-HTTP broker it can already reach —
|
||||
// forwards them. Best-effort: a dead log sink must never block an install, so
|
||||
// every outcome after authentication is 202.
|
||||
func (s *Server) handleLogs(w http.ResponseWriter, r *http.Request) {
|
||||
if s.vlinsertURL == "" || s.provisionToken == "" {
|
||||
http.Error(w, "log relay disabled: no vlinsert URL or no token configured", http.StatusServiceUnavailable)
|
||||
s.metrics.logRelay.WithLabelValues("disabled").Inc()
|
||||
return
|
||||
}
|
||||
if subtle.ConstantTimeCompare([]byte(bearer(r)), []byte(s.provisionToken)) != 1 {
|
||||
http.Error(w, "invalid or missing provision token", http.StatusUnauthorized)
|
||||
s.metrics.logRelay.WithLabelValues("unauthorized").Inc()
|
||||
return
|
||||
}
|
||||
body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxLogBody))
|
||||
if err != nil {
|
||||
http.Error(w, "log batch unreadable or larger than 1MiB", http.StatusBadRequest)
|
||||
s.metrics.logRelay.WithLabelValues("error").Inc()
|
||||
return
|
||||
}
|
||||
if len(bytes.TrimSpace(body)) == 0 {
|
||||
http.Error(w, "empty log batch", http.StatusBadRequest)
|
||||
s.metrics.logRelay.WithLabelValues("error").Inc()
|
||||
return
|
||||
}
|
||||
|
||||
lines := bytes.Count(bytes.TrimRight(body, "\n"), []byte("\n")) + 1
|
||||
if err := s.relayLogs(r.Context(), body, s.observedFields(r)); err != nil {
|
||||
slog.Error("log relay to vlinsert failed; dropping batch", "lines", lines, "err", err)
|
||||
s.metrics.logRelay.WithLabelValues("error").Inc()
|
||||
} else {
|
||||
s.metrics.logRelay.WithLabelValues("ok").Inc()
|
||||
s.metrics.logLines.Add(float64(lines))
|
||||
}
|
||||
s.ok(w, http.StatusAccepted, "text/plain", []byte("accepted\n"), "logs")
|
||||
}
|
||||
|
||||
// relayLogs makes one short-timeout POST to vlinsert's jsonline endpoint.
|
||||
// serial and phase are constant for a run and low-cardinality, so they key the
|
||||
// stream; MAC is per-NIC and rides in extra_fields so it stays searchable
|
||||
// without multiplying streams.
|
||||
func (s *Server) relayLogs(ctx context.Context, body []byte, extra string) error {
|
||||
q := url.Values{
|
||||
"_stream_fields": {"serial,phase"},
|
||||
"_msg_field": {"msg"},
|
||||
"_time_field": {"time"},
|
||||
}
|
||||
if extra != "" {
|
||||
q.Set("extra_fields", extra)
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.vlinsertURL+"/insert/jsonline?"+q.Encode(), bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/x-ndjson")
|
||||
resp, err := s.vlClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
_, _ = io.Copy(io.Discard, resp.Body)
|
||||
if resp.StatusCode >= http.StatusMultipleChoices {
|
||||
return fmt.Errorf("vlinsert: %s", resp.Status)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// observedFields builds the extra_fields value from what bootapi observes
|
||||
// rather than what the client claims: the source IP, plus the NetBox device
|
||||
// name when ?mac= resolves. Resolution is optional — a miss just omits it.
|
||||
func (s *Server) observedFields(r *http.Request) string {
|
||||
fields := []string{}
|
||||
if ip, _, err := net.SplitHostPort(r.RemoteAddr); err == nil && ip != "" {
|
||||
fields = append(fields, "src_ip="+ip)
|
||||
}
|
||||
if mac := r.URL.Query().Get("mac"); looksLikeMAC(mac) {
|
||||
if host, err := s.lookup(r.Context(), "mac", mac); err == nil {
|
||||
fields = append(fields, "device="+host.Hostname)
|
||||
}
|
||||
}
|
||||
return strings.Join(fields, ",")
|
||||
}
|
||||
|
||||
// bearer extracts a token from "Authorization: Bearer <t>" or a bare "token"
|
||||
// header.
|
||||
func bearer(r *http.Request) string {
|
||||
|
||||
Reference in New Issue
Block a user