unkin-agent
04ec9e6d41
Report the accumulated warnings on stderr when the estate load fails, and fold the dropped-record tally into the error. Count records rather than warning messages in --explain, by carrying the count on the Warn hook.
277 lines
8.2 KiB
Go
277 lines
8.2 KiB
Go
// Package puppetdb reads node facts from PuppetDB's query API. It is GET-only
|
|
// and unauthenticated — deliberately a separate client from the mTLS
|
|
// puppetserver one.
|
|
package puppetdb
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/url"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// ErrNodeNotFound is returned when PuppetDB knows no facts for a certname.
|
|
var ErrNodeNotFound = errors.New("node not found in PuppetDB")
|
|
|
|
// Client queries PuppetDB.
|
|
type Client struct {
|
|
BaseURL string
|
|
HTTP *http.Client
|
|
|
|
// Warn receives one message per batch of dropped fact records, with the
|
|
// number of records that batch accounts for — batches differ in size, so
|
|
// counting messages is not counting records. A nil Warn reports to stderr
|
|
// rather than discarding: a dropped record must never be silent. It is
|
|
// called on the goroutine that made the request, so a Client shared across
|
|
// goroutines needs a Warn that is safe to call concurrently.
|
|
Warn func(msg string, records int)
|
|
}
|
|
|
|
// New returns a client for baseURL.
|
|
func New(baseURL string, timeout time.Duration) *Client {
|
|
return &Client{
|
|
BaseURL: strings.TrimRight(baseURL, "/"),
|
|
HTTP: &http.Client{Timeout: timeout},
|
|
}
|
|
}
|
|
|
|
// factRecord is one fact as the facts endpoint returns it. Value stays raw so
|
|
// that a single undecodable fact can be dropped without costing the rest.
|
|
type factRecord struct {
|
|
Certname string `json:"certname"`
|
|
Name string `json:"name"`
|
|
Value json.RawMessage `json:"value"`
|
|
}
|
|
|
|
// FactRecord is one (certname, fact) pair with its value decoded.
|
|
type FactRecord struct {
|
|
Certname string
|
|
Name string
|
|
Value any
|
|
}
|
|
|
|
// Facts returns the node's facts as the values map a catalog request needs.
|
|
func (c *Client) Facts(ctx context.Context, certname string) (map[string]any, error) {
|
|
body, status, err := c.queryFacts(ctx, []string{"=", "certname", certname}, certname)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if status == http.StatusNotFound {
|
|
return nil, fmt.Errorf("%w: %s", ErrNodeNotFound, certname)
|
|
}
|
|
if status != http.StatusOK {
|
|
return nil, fmt.Errorf("puppetdb returned %d for %s facts: %s", status, certname, snippet(body))
|
|
}
|
|
|
|
records, err := decodeFacts(body, certname)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(records) == 0 {
|
|
return nil, fmt.Errorf("%w: %s", ErrNodeNotFound, certname)
|
|
}
|
|
return c.fold(certname, records)
|
|
}
|
|
|
|
// FactsByName returns the named facts for every node PuppetDB knows about, in
|
|
// a single query. Tools that evaluate selectors over the whole estate want the
|
|
// table once rather than a request per host.
|
|
func (c *Client) FactsByName(ctx context.Context, names []string) ([]FactRecord, error) {
|
|
if len(names) == 0 {
|
|
return nil, nil
|
|
}
|
|
clauses := make([]any, 0, len(names)+1)
|
|
clauses = append(clauses, "or")
|
|
for _, n := range names {
|
|
clauses = append(clauses, []any{"=", "name", n})
|
|
}
|
|
query := []any{"extract", []string{"certname", "name", "value"}, clauses}
|
|
|
|
body, status, err := c.queryFacts(ctx, query, "the estate")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if status != http.StatusOK {
|
|
return nil, fmt.Errorf("puppetdb returned %d for fact query: %s", status, snippet(body))
|
|
}
|
|
|
|
records, err := decodeRecords(body, "the estate")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
out := make([]FactRecord, 0, len(records))
|
|
var skipped []string
|
|
for _, r := range records {
|
|
v, ok := decodeValue(r)
|
|
if !ok {
|
|
skipped = append(skipped, r.Name)
|
|
continue
|
|
}
|
|
out = append(out, FactRecord{Certname: r.Certname, Name: r.Name, Value: v})
|
|
}
|
|
if len(skipped) > 0 {
|
|
c.warn(len(skipped), "skipped %d undecodable fact(s) in the estate fact table: %s",
|
|
len(skipped), strings.Join(skipped, ", "))
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// queryFacts runs an AST query against the facts endpoint. subject names what
|
|
// was asked for so a failure says so.
|
|
func (c *Client) queryFacts(ctx context.Context, query any, subject string) ([]byte, int, error) {
|
|
encoded, err := json.Marshal(query)
|
|
if err != nil {
|
|
return nil, 0, fmt.Errorf("building facts query for %s: %w", subject, err)
|
|
}
|
|
return c.get(ctx, "/pdb/query/v4/facts", string(encoded))
|
|
}
|
|
|
|
// decodeRecords reads the flat array of fact records the facts endpoint
|
|
// answers with.
|
|
func decodeRecords(body []byte, subject string) ([]factRecord, error) {
|
|
body = bytes.TrimPrefix(body, []byte("\xef\xbb\xbf"))
|
|
|
|
var records []factRecord
|
|
if err := json.Unmarshal(body, &records); err != nil {
|
|
return nil, fmt.Errorf("parsing facts for %s: %w", subject, err)
|
|
}
|
|
return records, nil
|
|
}
|
|
|
|
// decodeValue reports a record's value and whether it decoded at all.
|
|
func decodeValue(r factRecord) (any, bool) {
|
|
if len(r.Value) == 0 {
|
|
return nil, false
|
|
}
|
|
var v any
|
|
if json.Unmarshal(r.Value, &v) != nil {
|
|
return nil, false
|
|
}
|
|
return v, true
|
|
}
|
|
|
|
// decodeFacts reads the fact records a single-node query returns. A record
|
|
// belonging to another node is never substituted for the one that was asked
|
|
// for. A record that carries no certname of its own cannot be contradicting
|
|
// the query that named one, so it is kept.
|
|
func decodeFacts(body []byte, certname string) ([]factRecord, error) {
|
|
records, err := decodeRecords(body, certname)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
kept := make([]factRecord, 0, len(records))
|
|
for _, r := range records {
|
|
if r.Certname == "" || sameCertname(r.Certname, certname) {
|
|
kept = append(kept, r)
|
|
}
|
|
}
|
|
if len(kept) == 0 && len(records) > 0 {
|
|
return nil, fmt.Errorf("no facts for %s: puppetdb returned %s",
|
|
certname, strings.Join(certnames(records), ", "))
|
|
}
|
|
return kept, nil
|
|
}
|
|
|
|
// sameCertname compares certnames case-insensitively: they are DNS names, so
|
|
// folding can only absorb a certname that was normalised for case, never mask
|
|
// a different node.
|
|
func sameCertname(got, want string) bool {
|
|
return got != "" && strings.EqualFold(got, want)
|
|
}
|
|
|
|
// certnames lists the distinct nodes a response spoke for, in the order they
|
|
// appeared, so a mismatch error names what actually came back.
|
|
func certnames(records []factRecord) []string {
|
|
seen := make(map[string]bool, len(records))
|
|
names := make([]string, 0, len(records))
|
|
for _, r := range records {
|
|
name := r.Certname
|
|
if name == "" {
|
|
name = "(unnamed)"
|
|
}
|
|
if seen[name] {
|
|
continue
|
|
}
|
|
seen[name] = true
|
|
names = append(names, name)
|
|
}
|
|
return names
|
|
}
|
|
|
|
func (c *Client) get(ctx context.Context, path, query string) ([]byte, int, error) {
|
|
target := c.BaseURL + path + "?" + url.Values{"query": {query}}.Encode()
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
req.Header.Set("Accept", "application/json")
|
|
|
|
resp, err := c.HTTP.Do(req)
|
|
if err != nil {
|
|
return nil, 0, fmt.Errorf("puppetdb request failed: %w", err)
|
|
}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
|
|
body, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return nil, resp.StatusCode, fmt.Errorf("reading puppetdb response: %w", err)
|
|
}
|
|
return body, resp.StatusCode, nil
|
|
}
|
|
|
|
// fold turns fact records into the flat name -> value map the catalog API
|
|
// wants. Structured facts keep their nested shape. An undecodable record is
|
|
// dropped and reported: one bad fact must not cost the whole set. A response
|
|
// in which nothing decodes is a failed fetch, not an empty fact set.
|
|
func (c *Client) fold(certname string, records []factRecord) (map[string]any, error) {
|
|
values := make(map[string]any, len(records))
|
|
var skipped []string
|
|
for _, r := range records {
|
|
if r.Name == "" {
|
|
continue
|
|
}
|
|
v, ok := decodeValue(r)
|
|
if !ok {
|
|
skipped = append(skipped, r.Name)
|
|
continue
|
|
}
|
|
values[r.Name] = v
|
|
}
|
|
if len(values) == 0 {
|
|
return nil, fmt.Errorf("no usable facts for %s: all %d records were undecodable", certname, len(records))
|
|
}
|
|
if len(skipped) > 0 {
|
|
c.warn(len(skipped), "skipped %d undecodable fact(s) for %s: %s",
|
|
len(skipped), certname, strings.Join(skipped, ", "))
|
|
}
|
|
return values, nil
|
|
}
|
|
|
|
// warn reports records dropped. An unwired hook falls back to stderr so that
|
|
// forgetting to set one loses visibility, never the record itself.
|
|
func (c *Client) warn(records int, format string, args ...any) {
|
|
msg := fmt.Sprintf(format, args...)
|
|
if c.Warn != nil {
|
|
c.Warn(msg, records)
|
|
return
|
|
}
|
|
fmt.Fprintln(os.Stderr, "warning:", msg)
|
|
}
|
|
|
|
func snippet(body []byte) string {
|
|
s := strings.TrimSpace(string(body))
|
|
if len(s) > 200 {
|
|
return s[:200] + "..."
|
|
}
|
|
return s
|
|
}
|