From daf3fe846290dfdd49144e6449d921500b7ed01f Mon Sep 17 00:00:00 2001 From: Philippe Scorsolini Date: Thu, 22 Jan 2026 11:36:56 +0100 Subject: [PATCH 1/2] feat: initial implementation Signed-off-by: Philippe Scorsolini --- Dockerfile | 45 +++++ README.md | 83 +++++++-- go.mod | 26 +++ go.sum | 72 ++++++++ main.go | 136 +++++++++++++++ server/server.go | 176 +++++++++++++++++++ server/server_test.go | 394 ++++++++++++++++++++++++++++++++++++++++++ 7 files changed, 915 insertions(+), 17 deletions(-) create mode 100644 Dockerfile create mode 100644 go.mod create mode 100644 go.sum create mode 100644 main.go create mode 100644 server/server.go create mode 100644 server/server_test.go diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..a4be505 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,45 @@ +# syntax=docker/dockerfile:1 + +# We use the latest Go 1.x version unless asked to use something else. +# The GitHub Actions CI job sets this argument for a consistent Go version. +ARG GO_VERSION=1 + +# Setup the base environment. The BUILDPLATFORM is set automatically by Docker. +# The --platform=${BUILDPLATFORM} flag tells Docker to build the function using +# the OS and architecture of the host running the build, not the OS and +# architecture that we're building the function for. +FROM --platform=${BUILDPLATFORM} golang:${GO_VERSION} AS build + +WORKDIR /inspector + +# We don't want or need CGo support, so we disable it. +ENV CGO_ENABLED=0 + +# We run go mod download in a separate step so that we can cache its results. +# This lets us avoid re-downloading modules if we don't need to. The type=target +# mount tells Docker to mount the current directory read-only in the WORKDIR. +# The type=cache mount tells Docker to cache the Go modules cache across builds. +RUN --mount=target=. --mount=type=cache,target=/go/pkg/mod go mod download + +# The TARGETOS and TARGETARCH args are set by docker. We set GOOS and GOARCH to +# these values to ask Go to compile a binary for these architectures. If +# TARGETOS and TARGETOS are different from BUILDPLATFORM, Go will cross compile +# for us (e.g. compile a linux/amd64 binary on a linux/arm64 build machine). +ARG TARGETOS +ARG TARGETARCH + +# Build the main binary. The type=target mount tells Docker to mount the +# current directory read-only in the WORKDIR. The type=cache mount tells Docker +# to cache the Go modules cache across builds. +RUN --mount=target=. \ + --mount=type=cache,target=/go/pkg/mod \ + --mount=type=cache,target=/root/.cache/go-build \ + GOOS=${TARGETOS} GOARCH=${TARGETARCH} go build -o /inspector-sidecar . + +# Produce the Function image. We use a very lightweight 'distroless' image that +# does not include any of the build tools used in previous stages. +FROM gcr.io/distroless/static-debian12:nonroot AS image +WORKDIR / +COPY --from=build /inspector-sidecar /inspector-sidecar +USER nonroot:nonroot +ENTRYPOINT ["/inspector-sidecar"] diff --git a/README.md b/README.md index 5094e8c..cec4f01 100644 --- a/README.md +++ b/README.md @@ -1,23 +1,37 @@ # Pipeline Inspector Sidecar -This repo is a small and streamlined implementation of a sidecar container for Crossplane, -capturing Functions' Requests and Responses and printing them to the pod logs. +A minimal reference implementation of a sidecar container for Crossplane that captures +function pipeline execution data (requests and responses) and logs them to stdout. The full design of this feature can be found in the [design doc](https://github.com/crossplane/crossplane/blob/main/design/one-pager-pipeline-inspector.md). +## Features + +- Captures `RunFunctionRequest` and `RunFunctionResponse` data for each function invocation +- Supports JSON and human-readable text output formats +- Correlates pipeline steps using trace IDs, span IDs, and step indices +- Runs as a non-root user in a minimal distroless container + +## CLI Flags + +| Flag | Environment Variable | Default | Description | +|------|---------------------|---------|-------------| +| `--socket-path` | `PIPELINE_INSPECTOR_SOCKET` | `/var/run/pipeline-inspector/socket` | Unix socket path to listen on | +| `--format` | - | `json` | Output format (`json` or `text`) | +| `--max-recv-msg-size` | `MAX_RECV_MSG_SIZE` | `4194304` (4MB) | Maximum gRPC receive message size in bytes | +| `--shutdown-timeout` | `SHUTDOWN_TIMEOUT` | `5s` | Graceful shutdown timeout | + ## Usage This repository publishes release images to -`xpkg.crossplane.io/crossplane/inspector-sidecar`. This image can then be -included as a sidecar container for Crossplane through the Helm chart's -values. +`xpkg.crossplane.io/crossplane/inspector-sidecar`. This image can be +included as a sidecar container for Crossplane through the Helm chart's values. ```yaml # Example: # helm upgrade --install crossplane crossplane/crossplane \ # -n crossplane-system --create-namespace \ -# -f pipeline-inspector-values.yaml \ -# --set image.tag=v0.0.0-hack +# -f pipeline-inspector-values.yaml args: - --enable-pipeline-inspector @@ -33,9 +47,11 @@ extraVolumeMountsCrossplane: sidecarsCrossplane: - name: pipeline-inspector - image: crossplane/inspector-sidecar - command: - - /usr/local/bin/pipeline-inspector + image: xpkg.crossplane.io/crossplane/inspector-sidecar:latest + args: + - --format=json + # Increase if your function payloads exceed 4MB + # - --max-recv-msg-size=8388608 volumeMounts: - name: pipeline-inspector-socket mountPath: /var/run/pipeline-inspector @@ -46,14 +62,47 @@ sidecarsCrossplane: limits: cpu: 100m memory: 128Mi +``` +## Output Formats + +### JSON Format (default) + +```json +{"meta":{"compositeResourceApiVersion":"example.org/v1","compositeResourceKind":"XDatabase","compositeResourceName":"my-db","compositeResourceNamespace":"default","compositeResourceUid":"abc-123","compositionName":"my-composition","functionName":"function-patch-and-transform","iteration":0,"spanId":"span-456","stepIndex":0,"timestamp":"2026-01-15T10:30:00Z","traceId":"trace-789"},"payload":{...},"type":"REQUEST"} ``` -When this container starts up, it starts a gRPC server that listens on a unix -domain socket at the default path of `/var/run/pipeline-inspector/socket`, -which Crossplane is going to send RunFunctionRequests and RunFunctionResponses -from Functions. +### Text Format -The gRPC server implementation in this repo accepts incoming payloads -and simply writes them to `stdout` so they will be included in the provider -pod's logs. +Use `--format=text` for human-readable output: + +``` +=== REQUEST === + XR: example.org/v1/XDatabase (my-db) + XR UID: abc-123 + XR NS: default + Composition: my-composition + Function: function-patch-and-transform (step 0, iteration 0) + Trace ID: trace-789 + Span ID: span-456 + Timestamp: 2026-01-15T10:30:00.000Z + Payload: + apiVersion: apiextensions.crossplane.io/v1 + ... +``` + +## Building + +```bash +# Build locally +go build -o inspector-sidecar . + +# Build Docker image +docker build -t inspector-sidecar . +``` + +## Testing + +```bash +go test ./... +``` diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..ce3d661 --- /dev/null +++ b/go.mod @@ -0,0 +1,26 @@ +module github.com/crossplane/inspector-sidecar + +go 1.24.9 + +replace github.com/crossplane/crossplane-runtime/v2 => github.com/phisco/crossplane-runtime/v2 v2.0.0-20260122083359-147d66f5fee9 + +require ( + github.com/alecthomas/kong v1.10.0 + github.com/crossplane/crossplane-runtime/v2 v2.0.0 + github.com/go-logr/zapr v1.3.0 + go.uber.org/zap v1.27.1 + google.golang.org/grpc v1.75.1 + google.golang.org/protobuf v1.36.11 + sigs.k8s.io/yaml v1.6.0 +) + +require ( + github.com/go-logr/logr v1.4.3 // indirect + go.uber.org/multierr v1.10.0 // indirect + go.yaml.in/yaml/v2 v2.4.2 // indirect + golang.org/x/net v0.43.0 // indirect + golang.org/x/sys v0.35.0 // indirect + golang.org/x/text v0.28.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20250707201910-8d1bb00bc6a7 // indirect + k8s.io/klog/v2 v2.130.1 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..52809ad --- /dev/null +++ b/go.sum @@ -0,0 +1,72 @@ +github.com/alecthomas/assert/v2 v2.11.0 h1:2Q9r3ki8+JYXvGsDyBXwH3LcJ+WK5D0gc5E8vS6K3D0= +github.com/alecthomas/assert/v2 v2.11.0/go.mod h1:Bze95FyfUr7x34QZrjL+XP+0qgp/zg8yS+TtBj1WA3k= +github.com/alecthomas/kong v1.10.0 h1:8K4rGDpT7Iu+jEXCIJUeKqvpwZHbsFRoebLbnzlmrpw= +github.com/alecthomas/kong v1.10.0/go.mod h1:p2vqieVMeTAnaC83txKtXe8FLke2X07aruPWXyMPQrU= +github.com/alecthomas/repr v0.4.0 h1:GhI2A8MACjfegCPVq9f1FLvIBS+DrQ2KQBFZP1iFzXc= +github.com/alecthomas/repr v0.4.0/go.mod h1:Fr0507jx4eOXV7AlPV6AVZLYrLIuIeSOWtW57eE/O/4= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/go-logr/zapr v1.3.0 h1:XGdV8XW8zdwFiwOA2Dryh1gj2KRQyOOoNmBy4EplIcQ= +github.com/go-logr/zapr v1.3.0/go.mod h1:YKepepNBd1u/oyhd/yQmtjVXmm9uML4IXUgMOwR8/Gg= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUqJM= +github.com/hexops/gotextdiff v1.0.3/go.mod h1:pSWU5MAI3yDq+fZBTazCSJysOMbxWL1BSow5/V2vxeg= +github.com/phisco/crossplane-runtime/v2 v2.0.0-20260122083359-147d66f5fee9 h1:GtDHcfGpjshxMf+ByWR8RkEmXPwSQXUxOAx/RKZxLt4= +github.com/phisco/crossplane-runtime/v2 v2.0.0-20260122083359-147d66f5fee9/go.mod h1:WVVus9FBbAVjAmFxrOGDdZBFuUv9TqR916JmVl3PVRk= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= +go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A= +go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ= +go.opentelemetry.io/otel v1.37.0/go.mod h1:ehE/umFRLnuLa/vSccNq9oS1ErUlkkK71gMcN34UG8I= +go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/WgbsdpcPoZE= +go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E= +go.opentelemetry.io/otel/sdk v1.37.0 h1:ItB0QUqnjesGRvNcmAcU0LyvkVyGJ2xftD29bWdDvKI= +go.opentelemetry.io/otel/sdk v1.37.0/go.mod h1:VredYzxUvuo2q3WRcDnKDjbdvmO0sCzOvVAiY+yUkAg= +go.opentelemetry.io/otel/sdk/metric v1.37.0 h1:90lI228XrB9jCMuSdA0673aubgRobVZFhbjxHHspCPc= +go.opentelemetry.io/otel/sdk/metric v1.37.0/go.mod h1:cNen4ZWfiD37l5NhS+Keb5RXVWZWpRE+9WyVCpbo5ps= +go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4= +go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ= +go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= +go.uber.org/zap v1.27.1 h1:08RqriUEv8+ArZRYSTXy1LeBScaMpVSTBhCeaZYfMYc= +go.uber.org/zap v1.27.1/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= +go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= +go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= +go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= +go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE= +golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg= +golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= +golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng= +golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU= +gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= +gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20250707201910-8d1bb00bc6a7 h1:pFyd6EwwL2TqFf8emdthzeX+gZE1ElRq3iM8pui4KBY= +google.golang.org/genproto/googleapis/rpc v0.0.0-20250707201910-8d1bb00bc6a7/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A= +google.golang.org/grpc v1.75.1 h1:/ODCNEuf9VghjgO3rqLcfg8fiOP0nSluljWFlDxELLI= +google.golang.org/grpc v1.75.1/go.mod h1:JtPAzKiq4v1xcAB2hydNlWI2RnF85XXcV0mhKXr2ecQ= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk= +k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE= +sigs.k8s.io/yaml v1.6.0 h1:G8fkbMSAFqgEFgh4b1wmtzDnioxFCUgTZhlbj5P9QYs= +sigs.k8s.io/yaml v1.6.0/go.mod h1:796bPqUfzR/0jLAl6XjHl3Ck7MiyVv8dbTdyT3/pMf4= diff --git a/main.go b/main.go new file mode 100644 index 0000000..877548f --- /dev/null +++ b/main.go @@ -0,0 +1,136 @@ +/* +Copyright 2026 The Crossplane Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); you may not use +this file except in compliance with the License. You may obtain a copy of the +License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software distributed +under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR +CONDITIONS OF ANY KIND, either express or implied. See the License for the +specific language governing permissions and limitations under the License. +*/ + +// Package main implements a reference Pipeline Inspector sidecar that logs +// function pipeline execution data to stdout. +package main + +import ( + "context" + "fmt" + "net" + "os" + "os/signal" + "syscall" + "time" + + "github.com/alecthomas/kong" + "github.com/go-logr/zapr" + "go.uber.org/zap" + "google.golang.org/grpc" + + pipelinev1alpha1 "github.com/crossplane/crossplane-runtime/v2/apis/pipelineinspector/proto/v1alpha1" + "github.com/crossplane/crossplane-runtime/v2/pkg/logging" + "github.com/crossplane/inspector-sidecar/server" +) + +// CLI arguments. +type CLI struct { + Debug bool `help:"Emit debug logs in addition to info logs." short:"d"` + SocketPath string `default:"/var/run/pipeline-inspector/socket" env:"PIPELINE_INSPECTOR_SOCKET" help:"Unix socket path to listen on."` + Format string `default:"json" enum:"json,text" help:"Output format (json or text)."` + MaxRecvMsgSize int `default:"4194304" env:"MAX_RECV_MSG_SIZE" help:"Maximum gRPC receive message size in bytes (default 4MB)."` + ShutdownTimeout time.Duration `default:"5s" env:"SHUTDOWN_TIMEOUT" help:"Graceful shutdown timeout."` +} + +func main() { + var cli CLI + kong.Parse(&cli) + + if err := run(cli); err != nil { + fmt.Fprintf(os.Stderr, "error: %v\n", err) + os.Exit(1) + } +} + +func run(cli CLI) error { + // Create logger. + log, err := newLogger(cli.Debug) + if err != nil { + return fmt.Errorf("cannot create logger: %w", err) + } + + // Remove existing socket file if it exists. + if err := os.Remove(cli.SocketPath); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("cannot remove existing socket: %w", err) + } + + // Listen on Unix socket. + lc := net.ListenConfig{} + listener, err := lc.Listen(context.Background(), "unix", cli.SocketPath) + if err != nil { + return fmt.Errorf("cannot listen on socket: %w", err) + } + defer func() { _ = listener.Close() }() + + log.Info("Pipeline Inspector listening", "socket", cli.SocketPath, "format", cli.Format) + + // Create gRPC server. + grpcServer := grpc.NewServer(grpc.MaxRecvMsgSize(cli.MaxRecvMsgSize)) + inspector := server.NewInspector(cli.Format, server.WithLogger(log)) + pipelinev1alpha1.RegisterPipelineInspectorServiceServer(grpcServer, inspector) + + // Handle shutdown signals. + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer cancel() + + go func() { + <-ctx.Done() + log.Info("Shutting down") + + // Create a timeout context for graceful shutdown. + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), cli.ShutdownTimeout) + defer shutdownCancel() + + // Try graceful shutdown first. + stopped := make(chan struct{}) + go func() { + grpcServer.GracefulStop() + close(stopped) + }() + + select { + case <-shutdownCtx.Done(): + log.Info("Graceful shutdown timed out, forcing stop") + grpcServer.Stop() + case <-stopped: + // Graceful shutdown completed. + } + }() + + // Serve requests. + if err := grpcServer.Serve(listener); err != nil { + return fmt.Errorf("server error: %w", err) + } + + return nil +} + +// newLogger creates a new logger based on the debug flag. +func newLogger(debug bool) (logging.Logger, error) { + var zl *zap.Logger + var err error + + if debug { + zl, err = zap.NewDevelopment() + } else { + zl, err = zap.NewProduction() + } + if err != nil { + return nil, err + } + + return logging.NewLogrLogger(zapr.NewLogger(zl)), nil +} diff --git a/server/server.go b/server/server.go new file mode 100644 index 0000000..2b1be99 --- /dev/null +++ b/server/server.go @@ -0,0 +1,176 @@ +/* +Copyright 2026 The Crossplane Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); you may not use +this file except in compliance with the License. You may obtain a copy of the +License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software distributed +under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR +CONDITIONS OF ANY KIND, either express or implied. See the License for the +specific language governing permissions and limitations under the License. +*/ + +// Package server implements a gRPC server for the Pipeline Inspector. +package server + +import ( + "context" + "encoding/json" + "fmt" + "io" + "os" + "strings" + + "google.golang.org/protobuf/encoding/protojson" + "sigs.k8s.io/yaml" + + pipelinev1alpha1 "github.com/crossplane/crossplane-runtime/v2/apis/pipelineinspector/proto/v1alpha1" + "github.com/crossplane/crossplane-runtime/v2/pkg/logging" +) + +// Inspector implements the PipelineInspectorService by logging to a writer. +type Inspector struct { + pipelinev1alpha1.UnimplementedPipelineInspectorServiceServer + + format string + out io.Writer + log logging.Logger +} + +// Option configures an Inspector. +type Option func(*Inspector) + +// WithOutput sets the output writer (default: os.Stdout). +func WithOutput(w io.Writer) Option { + return func(i *Inspector) { + i.out = w + } +} + +// WithLogger sets the logger for the Inspector. +func WithLogger(l logging.Logger) Option { + return func(i *Inspector) { + i.log = l + } +} + +// NewInspector creates a new Inspector with the given output format. +func NewInspector(format string, opts ...Option) *Inspector { + i := &Inspector{ + format: format, + out: os.Stdout, + log: logging.NewNopLogger(), + } + for _, opt := range opts { + opt(i) + } + return i +} + +// EmitRequest logs the function request before execution. +func (i *Inspector) EmitRequest(_ context.Context, req *pipelinev1alpha1.EmitRequestRequest) (*pipelinev1alpha1.EmitRequestResponse, error) { + // Decode JSON payload from bytes. + payload := decodeJSONPayload(req.GetRequest()) + i.logEvent("REQUEST", req.GetMeta(), payload, "") + return &pipelinev1alpha1.EmitRequestResponse{}, nil +} + +// EmitResponse logs the function response after execution. +func (i *Inspector) EmitResponse(_ context.Context, req *pipelinev1alpha1.EmitResponseRequest) (*pipelinev1alpha1.EmitResponseResponse, error) { + // Decode JSON payload from bytes. + payload := decodeJSONPayload(req.GetResponse()) + i.logEvent("RESPONSE", req.GetMeta(), payload, req.GetError()) + return &pipelinev1alpha1.EmitResponseResponse{}, nil +} + +// decodeJSONPayload decodes JSON bytes into a map for display. +func decodeJSONPayload(data []byte) any { + if len(data) == 0 { + return nil + } + var result any + if err := json.Unmarshal(data, &result); err != nil { + // If we can't decode, return the raw string. + return string(data) + } + return result +} + +func (i *Inspector) logEvent(eventType string, meta *pipelinev1alpha1.StepMeta, payload any, errMsg string) { + if i.format == "text" { + i.logText(eventType, meta, payload, errMsg) + return + } + i.logJSON(eventType, meta, payload, errMsg) +} + +func (i *Inspector) logJSON(eventType string, meta *pipelinev1alpha1.StepMeta, payload any, errMsg string) { + // Marshal meta using protojson to preserve proto field names. + metaJSON, err := protojson.Marshal(meta) + if err != nil { + i.log.Debug("Cannot marshal meta", "error", err) + return + } + + // Unmarshal meta into a map so we can include it in the final event. + var metaMap map[string]any + if err := json.Unmarshal(metaJSON, &metaMap); err != nil { + i.log.Debug("Cannot unmarshal meta", "error", err) + return + } + + event := map[string]any{ + "type": eventType, + "meta": metaMap, + "payload": payload, + } + if errMsg != "" { + event["error"] = errMsg + } + + eventJSON, err := json.Marshal(event) + if err != nil { + i.log.Debug("Cannot marshal event", "error", err) + return + } + + _, _ = fmt.Fprintln(i.out, string(eventJSON)) +} + +func (i *Inspector) logText(eventType string, meta *pipelinev1alpha1.StepMeta, payload any, errMsg string) { + _, _ = fmt.Fprintf(i.out, "=== %s ===\n", eventType) + _, _ = fmt.Fprintf(i.out, " XR: %s/%s (%s)\n", meta.GetCompositeResourceApiVersion(), meta.GetCompositeResourceKind(), meta.GetCompositeResourceName()) + _, _ = fmt.Fprintf(i.out, " XR UID: %s\n", meta.GetCompositeResourceUid()) + if ns := meta.GetCompositeResourceNamespace(); ns != "" { + _, _ = fmt.Fprintf(i.out, " XR NS: %s\n", ns) + } + _, _ = fmt.Fprintf(i.out, " Composition: %s\n", meta.GetCompositionName()) + _, _ = fmt.Fprintf(i.out, " Function: %s (step %d, iteration %d)\n", meta.GetFunctionName(), meta.GetStepIndex(), meta.GetIteration()) + _, _ = fmt.Fprintf(i.out, " Trace ID: %s\n", meta.GetTraceId()) + _, _ = fmt.Fprintf(i.out, " Span ID: %s\n", meta.GetSpanId()) + _, _ = fmt.Fprintf(i.out, " Timestamp: %s\n", meta.GetTimestamp().AsTime().Format("2006-01-02T15:04:05.000Z07:00")) + if errMsg != "" { + _, _ = fmt.Fprintf(i.out, " Error: %s\n", errMsg) + } + + // Pretty-print payload as YAML for readability. + if payload != nil { + payloadYAML, err := yaml.Marshal(payload) + if err == nil { + _, _ = fmt.Fprintf(i.out, " Payload:\n%s\n", indentLines(string(payloadYAML), " ")) + } + } + _, _ = fmt.Fprintln(i.out) +} + +// indentLines adds the given prefix to each line of the input string. +func indentLines(s, prefix string) string { + var result strings.Builder + for line := range strings.SplitSeq(strings.TrimSuffix(s, "\n"), "\n") { + result.WriteString(prefix + line + "\n") + } + return result.String() +} diff --git a/server/server_test.go b/server/server_test.go new file mode 100644 index 0000000..e049d4e --- /dev/null +++ b/server/server_test.go @@ -0,0 +1,394 @@ +/* +Copyright 2026 The Crossplane Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); you may not use +this file except in compliance with the License. You may obtain a copy of the +License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software distributed +under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR +CONDITIONS OF ANY KIND, either express or implied. See the License for the +specific language governing permissions and limitations under the License. +*/ + +package server + +import ( + "bytes" + "context" + "encoding/json" + "strings" + "testing" + "time" + + "google.golang.org/protobuf/types/known/timestamppb" + + pipelinev1alpha1 "github.com/crossplane/crossplane-runtime/v2/apis/pipelineinspector/proto/v1alpha1" +) + +func TestEmitRequest_JSON(t *testing.T) { + var buf bytes.Buffer + inspector := NewInspector("json", WithOutput(&buf)) + + meta := &pipelinev1alpha1.StepMeta{ + TraceId: "trace-123", + SpanId: "span-456", + StepIndex: 0, + Iteration: 0, + FunctionName: "function-patch-and-transform", + CompositionName: "my-composition", + CompositeResourceUid: "uid-789", + CompositeResourceName: "my-xr", + CompositeResourceNamespace: "default", + CompositeResourceApiVersion: "example.org/v1", + CompositeResourceKind: "XDatabase", + Timestamp: timestamppb.New(time.Now()), + } + + req := &pipelinev1alpha1.EmitRequestRequest{ + Request: []byte(`{"apiVersion":"apiextensions.crossplane.io/v1"}`), + Meta: meta, + } + + _, err := inspector.EmitRequest(context.Background(), req) + if err != nil { + t.Fatalf("EmitRequest failed: %v", err) + } + + output := buf.String() + if !strings.Contains(output, `"type":"REQUEST"`) { + t.Errorf("expected type REQUEST in output, got: %s", output) + } + if !strings.Contains(output, `"functionName":"function-patch-and-transform"`) { + t.Errorf("expected functionName in output, got: %s", output) + } + + // Verify it's valid JSON. + var result map[string]any + if err := json.Unmarshal([]byte(output), &result); err != nil { + t.Errorf("output is not valid JSON: %v", err) + } + + // Verify meta is included as a nested object. + if _, ok := result["meta"]; !ok { + t.Errorf("expected meta field in output, got: %s", output) + } +} + +func TestEmitResponse_JSON(t *testing.T) { + var buf bytes.Buffer + inspector := NewInspector("json", WithOutput(&buf)) + + meta := &pipelinev1alpha1.StepMeta{ + TraceId: "trace-123", + SpanId: "span-456", + FunctionName: "my-function", + Timestamp: timestamppb.New(time.Now()), + } + + req := &pipelinev1alpha1.EmitResponseRequest{ + Response: []byte(`{"desired":{}}`), + Error: "", + Meta: meta, + } + + _, err := inspector.EmitResponse(context.Background(), req) + if err != nil { + t.Fatalf("EmitResponse failed: %v", err) + } + + output := buf.String() + if !strings.Contains(output, `"type":"RESPONSE"`) { + t.Errorf("expected type RESPONSE in output, got: %s", output) + } + + // Verify it's valid JSON. + var result map[string]any + if err := json.Unmarshal([]byte(output), &result); err != nil { + t.Errorf("output is not valid JSON: %v", err) + } +} + +func TestEmitResponse_WithError(t *testing.T) { + var buf bytes.Buffer + inspector := NewInspector("json", WithOutput(&buf)) + + meta := &pipelinev1alpha1.StepMeta{ + FunctionName: "my-function", + Timestamp: timestamppb.New(time.Now()), + } + req := &pipelinev1alpha1.EmitResponseRequest{ + Response: nil, + Error: "function execution failed", + Meta: meta, + } + + _, _ = inspector.EmitResponse(context.Background(), req) + + output := buf.String() + if !strings.Contains(output, `"error":"function execution failed"`) { + t.Errorf("expected error field in output, got: %s", output) + } +} + +func TestEmitRequest_Text(t *testing.T) { + var buf bytes.Buffer + inspector := NewInspector("text", WithOutput(&buf)) + + meta := &pipelinev1alpha1.StepMeta{ + CompositeResourceApiVersion: "example.org/v1", + CompositeResourceKind: "XDatabase", + CompositeResourceName: "my-xr", + CompositeResourceUid: "uid-123", + CompositeResourceNamespace: "my-namespace", + CompositionName: "my-composition", + FunctionName: "my-function", + TraceId: "trace-abc", + SpanId: "span-def", + StepIndex: 1, + Iteration: 2, + Timestamp: timestamppb.New(time.Date(2026, 1, 15, 10, 30, 0, 0, time.UTC)), + } + + req := &pipelinev1alpha1.EmitRequestRequest{ + Request: []byte(`{"apiVersion":"apiextensions.crossplane.io/v1"}`), + Meta: meta, + } + + _, _ = inspector.EmitRequest(context.Background(), req) + + output := buf.String() + + // Verify header. + if !strings.Contains(output, "=== REQUEST ===") { + t.Errorf("expected REQUEST header, got: %s", output) + } + + // Verify XR info. + if !strings.Contains(output, "XR: example.org/v1/XDatabase (my-xr)") { + t.Errorf("expected XR info in output, got: %s", output) + } + + // Verify UID is included. + if !strings.Contains(output, "XR UID: uid-123") { + t.Errorf("expected XR UID in output, got: %s", output) + } + + // Verify namespace is included. + if !strings.Contains(output, "XR NS: my-namespace") { + t.Errorf("expected XR namespace in output, got: %s", output) + } + + // Verify composition name. + if !strings.Contains(output, "Composition: my-composition") { + t.Errorf("expected composition name in output, got: %s", output) + } + + // Verify function info. + if !strings.Contains(output, "Function: my-function (step 1, iteration 2)") { + t.Errorf("expected function info in output, got: %s", output) + } + + // Verify trace and span IDs. + if !strings.Contains(output, "Trace ID: trace-abc") { + t.Errorf("expected trace ID in output, got: %s", output) + } + if !strings.Contains(output, "Span ID: span-def") { + t.Errorf("expected span ID in output, got: %s", output) + } +} + +func TestEmitRequest_Text_NoNamespace(t *testing.T) { + var buf bytes.Buffer + inspector := NewInspector("text", WithOutput(&buf)) + + // Cluster-scoped resource has empty namespace. + meta := &pipelinev1alpha1.StepMeta{ + CompositeResourceApiVersion: "example.org/v1", + CompositeResourceKind: "XClusterDatabase", + CompositeResourceName: "my-cluster-xr", + CompositeResourceUid: "uid-456", + CompositeResourceNamespace: "", // Empty for cluster-scoped. + CompositionName: "cluster-composition", + FunctionName: "my-function", + Timestamp: timestamppb.New(time.Now()), + } + + req := &pipelinev1alpha1.EmitRequestRequest{ + Request: []byte(`{}`), + Meta: meta, + } + + _, _ = inspector.EmitRequest(context.Background(), req) + + output := buf.String() + + // Verify namespace line is NOT included for cluster-scoped resources. + if strings.Contains(output, "XR NS:") { + t.Errorf("expected no XR NS line for cluster-scoped resource, got: %s", output) + } +} + +func TestEmitResponse_Text_WithError(t *testing.T) { + var buf bytes.Buffer + inspector := NewInspector("text", WithOutput(&buf)) + + meta := &pipelinev1alpha1.StepMeta{ + FunctionName: "failing-function", + Timestamp: timestamppb.New(time.Now()), + } + + req := &pipelinev1alpha1.EmitResponseRequest{ + Response: nil, + Error: "something went wrong", + Meta: meta, + } + + _, _ = inspector.EmitResponse(context.Background(), req) + + output := buf.String() + if !strings.Contains(output, "Error: something went wrong") { + t.Errorf("expected error in output, got: %s", output) + } +} + +func TestDecodeJSONPayload(t *testing.T) { + tests := []struct { + name string + input []byte + wantNil bool + wantType string + }{ + { + name: "empty input", + input: nil, + wantNil: true, + }, + { + name: "valid json object", + input: []byte(`{"key":"value"}`), + wantType: "map[string]interface {}", + }, + { + name: "valid json array", + input: []byte(`[1,2,3]`), + wantType: "[]interface {}", + }, + { + name: "invalid json returns string", + input: []byte(`{not valid`), + wantType: "string", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := decodeJSONPayload(tt.input) + if tt.wantNil { + if result != nil { + t.Errorf("expected nil, got %v", result) + } + return + } + gotType := typeString(result) + if gotType != tt.wantType { + t.Errorf("expected type %s, got %s", tt.wantType, gotType) + } + }) + } +} + +func typeString(v any) string { + if v == nil { + return "" + } + switch v.(type) { + case map[string]any: + return "map[string]interface {}" + case []any: + return "[]interface {}" + case string: + return "string" + default: + return "unknown" + } +} + +func TestIndentLines(t *testing.T) { + tests := []struct { + name string + input string + prefix string + expected string + }{ + { + name: "single line", + input: "hello", + prefix: " ", + expected: " hello\n", + }, + { + name: "multiple lines", + input: "line1\nline2\nline3", + prefix: " ", + expected: " line1\n line2\n line3\n", + }, + { + name: "with trailing newline", + input: "line1\nline2\n", + prefix: " ", + expected: " line1\n line2\n", + }, + { + name: "empty string", + input: "", + prefix: " ", + expected: " \n", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := indentLines(tt.input, tt.prefix) + if result != tt.expected { + t.Errorf("indentLines(%q, %q) = %q, want %q", tt.input, tt.prefix, result, tt.expected) + } + }) + } +} + +func TestNewInspector_Defaults(t *testing.T) { + inspector := NewInspector("json") + + if inspector.format != "json" { + t.Errorf("expected format 'json', got %s", inspector.format) + } + if inspector.out == nil { + t.Error("expected out to be set") + } + if inspector.log == nil { + t.Error("expected log to be set") + } +} + +func TestNewInspector_WithOptions(t *testing.T) { + var out bytes.Buffer + inspector := NewInspector("text", WithOutput(&out)) + + if inspector.format != "text" { + t.Errorf("expected format 'text', got %s", inspector.format) + } + + // Verify custom writers are used. + inspector.EmitRequest(context.Background(), &pipelinev1alpha1.EmitRequestRequest{ + Meta: &pipelinev1alpha1.StepMeta{ + Timestamp: timestamppb.New(time.Now()), + }, + }) + + if out.Len() == 0 { + t.Error("expected output to be written to custom writer") + } +} From 68ff7887e20a4abf8178d1e94ddb705a68774572 Mon Sep 17 00:00:00 2001 From: Philippe Scorsolini Date: Tue, 27 Jan 2026 11:39:49 +0100 Subject: [PATCH 2/2] use upstream grpc interface Signed-off-by: Philippe Scorsolini --- go.mod | 4 +--- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/go.mod b/go.mod index ce3d661..cc8fc70 100644 --- a/go.mod +++ b/go.mod @@ -2,11 +2,9 @@ module github.com/crossplane/inspector-sidecar go 1.24.9 -replace github.com/crossplane/crossplane-runtime/v2 => github.com/phisco/crossplane-runtime/v2 v2.0.0-20260122083359-147d66f5fee9 - require ( github.com/alecthomas/kong v1.10.0 - github.com/crossplane/crossplane-runtime/v2 v2.0.0 + github.com/crossplane/crossplane-runtime/v2 v2.2.0-rc.0.0.20260127103424-627736f1b9f1 github.com/go-logr/zapr v1.3.0 go.uber.org/zap v1.27.1 google.golang.org/grpc v1.75.1 diff --git a/go.sum b/go.sum index 52809ad..e3a5577 100644 --- a/go.sum +++ b/go.sum @@ -4,6 +4,8 @@ github.com/alecthomas/kong v1.10.0 h1:8K4rGDpT7Iu+jEXCIJUeKqvpwZHbsFRoebLbnzlmrp github.com/alecthomas/kong v1.10.0/go.mod h1:p2vqieVMeTAnaC83txKtXe8FLke2X07aruPWXyMPQrU= github.com/alecthomas/repr v0.4.0 h1:GhI2A8MACjfegCPVq9f1FLvIBS+DrQ2KQBFZP1iFzXc= github.com/alecthomas/repr v0.4.0/go.mod h1:Fr0507jx4eOXV7AlPV6AVZLYrLIuIeSOWtW57eE/O/4= +github.com/crossplane/crossplane-runtime/v2 v2.2.0-rc.0.0.20260127103424-627736f1b9f1 h1:y5C/HJq9SXLYynEh4ZK6Tf7tDrsJNS+a5Ysjg9Eyc2I= +github.com/crossplane/crossplane-runtime/v2 v2.2.0-rc.0.0.20260127103424-627736f1b9f1/go.mod h1:WVVus9FBbAVjAmFxrOGDdZBFuUv9TqR916JmVl3PVRk= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= @@ -20,8 +22,6 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUqJM= github.com/hexops/gotextdiff v1.0.3/go.mod h1:pSWU5MAI3yDq+fZBTazCSJysOMbxWL1BSow5/V2vxeg= -github.com/phisco/crossplane-runtime/v2 v2.0.0-20260122083359-147d66f5fee9 h1:GtDHcfGpjshxMf+ByWR8RkEmXPwSQXUxOAx/RKZxLt4= -github.com/phisco/crossplane-runtime/v2 v2.0.0-20260122083359-147d66f5fee9/go.mod h1:WVVus9FBbAVjAmFxrOGDdZBFuUv9TqR916JmVl3PVRk= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=