nmon/internal/ingest/http.go

package ingest

import (
	"compress/gzip"
	"context"
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"strconv"
	"strings"
)

// Mount registers the HTTP ingest routes on mux:
//
//	POST /write                       Influx line protocol (v1 style)
//	POST /api/v1/write?format=influx  Influx line protocol
//	POST /api/v1/write                Prometheus remote write v1 (Content-Type: application/x-protobuf, Content-Encoding: snappy)
//
// Authentication is a push token only: "Authorization: Bearer <secret>", Influx's "Token <secret>", Basic auth
// with the secret as the password, or the v1 "p=<secret>" query parameter. Errors are RFC 9457 problem+json.
func (g *Ingester) Mount(mux *http.ServeMux) {
	mux.HandleFunc("POST /write", g.serveInflux)
	mux.HandleFunc("POST /api/v1/write", g.serveAPIWrite)
}

// Handler returns a handler with the routes of Mount, for tests and standalone listeners.
func (g *Ingester) Handler() http.Handler {
	mux := http.NewServeMux()
	g.Mount(mux)
	return mux
}

func (g *Ingester) serveAPIWrite(w http.ResponseWriter, r *http.Request) {
	ct, _, _ := strings.Cut(r.Header.Get("Content-Type"), ";")
	switch {
	case r.URL.Query().Get("format") == "influx":
		g.serveInflux(w, r)
	case strings.EqualFold(strings.TrimSpace(ct), "application/x-protobuf"):
		g.serveRemoteWrite(w, r)
	default:
		problem(w, http.StatusUnsupportedMediaType, "Unsupported media type",
			"POST /api/v1/write takes Prometheus remote write (application/x-protobuf with snappy) or ?format=influx")
	}
}

func (g *Ingester) serveInflux(w http.ResponseWriter, r *http.Request) {
	b, ok := g.authorize(w, r, "influx")
	if !ok {
		return
	}
	unit, err := precisionUnit(r.URL.Query().Get("precision"))
	if err != nil {
		problem(w, http.StatusBadRequest, "Bad precision", err.Error())
		return
	}
	body, ok := g.openBody(w, r, b)
	if !ok {
		return
	}
	defer func() { _ = body.Close() }()
	in, bad, err := parseInflux(body, unit)
	if err != nil {
		g.bodyError(w, b, err)
		return
	}
	g.drop(reasonParse, bad)
	g.finish(w, r.Context(), b, "influx", in, bad)
}

func (g *Ingester) serveRemoteWrite(w http.ResponseWriter, r *http.Request) {
	b, ok := g.authorize(w, r, "remote_write")
	if !ok {
		return
	}
	if enc := strings.ToLower(r.Header.Get("Content-Encoding")); enc != "snappy" {
		problem(w, http.StatusUnsupportedMediaType, "Unsupported encoding", "remote write needs Content-Encoding: snappy")
		return
	}
	r.Body = http.MaxBytesReader(w, r.Body, g.cfg.MaxBodyBytes)
	comp, err := io.ReadAll(r.Body)
	if err != nil {
		g.bodyError(w, b, err)
		return
	}
	pb, err := snappyDecode(comp, g.cfg.MaxBodyBytes)
	if err != nil {
		g.bodyError(w, b, err)
		return
	}
	in, err := parseRemoteWrite(pb)
	if err != nil {
		g.drop(reasonParse, 1)
		problem(w, http.StatusBadRequest, "Bad remote write body", err.Error())
		return
	}
	g.finish(w, r.Context(), b, "remote_write", in, 0)
}

// openBody caps the body and undoes gzip. The cap applies to the compressed bytes and again to the
// decompressed ones, so a gzip bomb stops at the same limit.
func (g *Ingester) openBody(w http.ResponseWriter, r *http.Request, b *Binding) (io.ReadCloser, bool) {
	limited := http.MaxBytesReader(w, r.Body, g.cfg.MaxBodyBytes)
	if !strings.EqualFold(r.Header.Get("Content-Encoding"), "gzip") {
		return limited, true
	}
	zr, err := gzip.NewReader(limited)
	if err != nil {
		g.bodyError(w, b, err)
		return nil, false
	}
	return &capReader{r: zr, left: g.cfg.MaxBodyBytes, c: limited}, true
}

// capReader fails with errTooLarge once left bytes have been read.
type capReader struct {
	r    io.Reader
	left int64
	c    io.Closer
}

func (c *capReader) Read(p []byte) (int, error) {
	if c.left <= 0 {
		return 0, errTooLarge
	}
	if int64(len(p)) > c.left {
		p = p[:c.left]
	}
	n, err := c.r.Read(p)
	c.left -= int64(n)
	return n, err
}

func (c *capReader) Close() error { return c.c.Close() }

// bodyError answers a body that could not be read: over the cap is 413 with a counter and an alarm.
func (g *Ingester) bodyError(w http.ResponseWriter, b *Binding, err error) {
	var mbe *http.MaxBytesError
	if errors.As(err, &mbe) || errors.Is(err, errTooLarge) {
		g.reject(b, reasonBodyCap, AlarmBodyCap, "request body over the "+strconv.FormatInt(g.cfg.MaxBodyBytes, 10)+" byte cap")
		problem(w, http.StatusRequestEntityTooLarge, "Body too large", "the body cap is "+strconv.FormatInt(g.cfg.MaxBodyBytes, 10)+" bytes")
		return
	}
	g.drop(reasonParse, 1)
	problem(w, http.StatusBadRequest, "Unreadable body", err.Error())
}

// finish ingests and maps the outcome: 204 when anything was accepted (X-Nmon-Dropped counts the rest),
// 429 when rate limited, 422 when everything was refused, 503 when the write queue failed.
func (g *Ingester) finish(w http.ResponseWriter, ctx context.Context, b *Binding, proto string, in []raw, bad int) {
	res, err := g.Ingest(ctx, b, proto, in)
	var rl *RateLimitedError
	switch {
	case errors.As(err, &rl):
		secs := int(rl.RetryAfter.Seconds()) + 1
		w.Header().Set("Retry-After", strconv.Itoa(secs))
		problem(w, http.StatusTooManyRequests, "Rate limit exceeded", rl.Error())
	case err != nil:
		problem(w, http.StatusServiceUnavailable, "Write queue unavailable", "the samples were not stored; retry")
	case res.Accepted == 0 && res.DroppedTotal()+bad > 0:
		problem(w, http.StatusUnprocessableEntity, "Nothing accepted", dropSummary(res, bad))
	default:
		if n := res.DroppedTotal() + bad; n > 0 {
			w.Header().Set("X-Nmon-Dropped", strconv.Itoa(n))
		}
		w.WriteHeader(http.StatusNoContent)
	}
}

func dropSummary(res Result, bad int) string {
	parts := make([]string, 0, len(res.Dropped)+1)
	for reason, n := range res.Dropped {
		parts = append(parts, reason+"="+strconv.Itoa(n))
	}
	if bad > 0 {
		parts = append(parts, reasonParse+"="+strconv.Itoa(bad))
	}
	return "dropped: " + strings.Join(parts, ", ")
}

// authorize extracts and verifies the push token. On failure it has written the response.
func (g *Ingester) authorize(w http.ResponseWriter, r *http.Request, proto string) (*Binding, bool) {
	secret := bearerSecret(r)
	b, err := g.cfg.Tokens.Authenticate(r.Context(), secret)
	switch {
	case err == nil:
		return b, true
	case errors.Is(err, ErrForbidden):
		problem(w, http.StatusForbidden, "Forbidden", "this token cannot push")
	case errors.Is(err, ErrUnauthenticated):
		g.reg.Counter("nmon_ingest_auth_failures_total", "proto", proto).Inc()
		w.Header().Set("WWW-Authenticate", `Bearer realm="nmon-push"`)
		problem(w, http.StatusUnauthorized, "Unauthorized", "a valid push token is required")
	default:
		problem(w, http.StatusInternalServerError, "Authentication unavailable", "try again")
	}
	return nil, false
}

func bearerSecret(r *http.Request) string {
	if h := r.Header.Get("Authorization"); h != "" {
		scheme, cred, _ := strings.Cut(h, " ")
		cred = strings.TrimSpace(cred)
		switch strings.ToLower(scheme) {
		case "bearer", "token":
			return cred
		case "basic":
			if _, pw, ok := r.BasicAuth(); ok {
				return pw
			}
		}
		return ""
	}
	return r.URL.Query().Get("p")
}

// problem writes an RFC 9457 problem document.
func problem(w http.ResponseWriter, status int, title, detail string) {
	w.Header().Set("Content-Type", "application/problem+json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(map[string]any{"type": "about:blank", "title": title, "status": status, "detail": detail})
}