Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 13 additions & 51 deletions cmd/crossplane/render/contextfn/listener.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,8 @@ package contextfn
import (
"context"
"encoding/json"
"fmt"
"net"
"os"
"path/filepath"
"sync"
"time"

Expand All @@ -37,16 +36,10 @@ import (

// Handle is the owner of a running in-process context function.
type Handle struct {
// Target is the gRPC target that dials the function. Set this as the
// FunctionInput address passed to the render engine.
Target string

target string
Comment thread
adamwg marked this conversation as resolved.
srv *grpc.Server
fn *server
socketPath string
dir string
stop sync.Once
log logging.Logger
seedInput *runtime.RawExtension
captureInput *runtime.RawExtension
}
Expand All @@ -58,9 +51,8 @@ func (h *Handle) Captured() *structpb.Struct {
}

// Start starts an in-process gRPC server that implements the composition
// function RunFunction RPC for context seeding and capture. The server
// listens on a unix-domain socket inside a fresh temp directory. Callers
// must call Handle.Stop when done.
// function RunFunction RPC for context seeding and capture. Callers must call
// Handle.Stop when done.
func Start(ctx context.Context, log logging.Logger, contextData map[string]any) (*Handle, error) {
si, err := json.Marshal(input{Mode: modeSeed})
if err != nil {
Expand All @@ -71,51 +63,24 @@ func Start(ctx context.Context, log logging.Logger, contextData map[string]any)
return nil, errors.Wrap(err, "cannot create capture context function input")
}

dir, err := os.MkdirTemp("", "render-ctx-*")
if err != nil {
return nil, errors.Wrap(err, "cannot create temp dir for context function socket")
}

cleanup := func() {
_ = os.RemoveAll(dir)
}

sockPath := filepath.Join(dir, "s")
var lc net.ListenConfig
lis, err := lc.Listen(ctx, "unix", sockPath)
lis, err := lc.Listen(ctx, "tcp", ":0")
if err != nil {
cleanup()
return nil, errors.Wrapf(err, "cannot listen on %q", sockPath)
}

cleanup = func() {
_ = lis.Close()
_ = os.RemoveAll(dir)
}

// In order for processes in Docker containers to connect to the socket, the
// socket must be world-writeable and its containing directory must be
// world-readable.
if err := os.Chmod(dir, 0o755); err != nil { //nolint:gosec // Necessary.
cleanup()
return nil, errors.Wrapf(err, "cannot make socket directory world-readable")
}
if err := os.Chmod(sockPath, 0o777); err != nil { //nolint:gosec // Necessary.
cleanup()
return nil, errors.Wrapf(err, "cannot make socket file writeable")
return nil, errors.Wrap(err, "cannot create listener for context function")
}

srv := grpc.NewServer(grpc.Creds(insecure.NewCredentials()))
fn := newServer(contextData)
fnv1.RegisterFunctionRunnerServiceServer(srv, fn)

addr := lis.Addr().(*net.TCPAddr) //nolint:forcetypeassert // We specified "tcp" above.

h := &Handle{
Target: "unix://" + sockPath,
// Report the target as 127.0.0.1:PORT since the render machinery knows
// how to handle functions listening on loopback.
target: fmt.Sprintf("127.0.0.1:%d", addr.Port),
Comment thread
coderabbitai[bot] marked this conversation as resolved.
srv: srv,
fn: fn,
socketPath: sockPath,
dir: dir,
log: log,
seedInput: &runtime.RawExtension{Raw: si},
captureInput: &runtime.RawExtension{Raw: ci},
}
Expand All @@ -129,8 +94,8 @@ func Start(ctx context.Context, log logging.Logger, contextData map[string]any)
return h, nil
}

// Stop gracefully stops the function server and removes the socket directory.
// Safe to call multiple times.
// Stop gracefully stops the function server, which closes its listener. Safe to
// call multiple times.
func (h *Handle) Stop() {
h.stop.Do(func() {
done := make(chan struct{})
Expand All @@ -143,8 +108,5 @@ func (h *Handle) Stop() {
case <-time.After(5 * time.Second):
h.srv.Stop()
}
if err := os.RemoveAll(h.dir); err != nil {
h.log.Debug("Cannot remove context function socket directory", "dir", h.dir, "error", err)
}
})
}
26 changes: 1 addition & 25 deletions cmd/crossplane/render/contextfn/listener_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@ package contextfn

import (
"context"
"os"
"testing"
"time"

Expand All @@ -42,7 +41,7 @@ func TestListenerRoundTrip(t *testing.T) {
}
defer h.Stop()

conn, err := grpc.NewClient(h.Target, grpc.WithTransportCredentials(insecure.NewCredentials()))
conn, err := grpc.NewClient(h.target, grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
t.Fatalf("grpc.NewClient: %v", err)
}
Expand All @@ -61,26 +60,3 @@ func TestListenerRoundTrip(t *testing.T) {
t.Errorf("context (-want +got):\n%s", diff)
}
}

func TestStopRemovesSocket(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

h, err := Start(ctx, logging.NewNopLogger(), nil)
if err != nil {
t.Fatalf("start: %v", err)
}

if _, err := os.Stat(h.socketPath); err != nil {
t.Fatalf("socket should exist: %v", err)
}

h.Stop()

if _, err := os.Stat(h.socketPath); !os.IsNotExist(err) {
t.Errorf("socket should not exist after Stop, got err=%v", err)
}

// Second Stop is a no-op.
h.Stop()
}
12 changes: 8 additions & 4 deletions cmd/crossplane/render/contextfn/wire.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,9 @@ import (
// Mirror the render package's runtime-selection annotations. Duplicated here
// rather than importing cmd/crank/render to avoid an import cycle.
const (
annotationKeyRuntime = "render.crossplane.io/runtime"
annotationValueRuntimeInProcess = "InProcess"
annotationKeyRuntime = "render.crossplane.io/runtime"
annotationValueRuntimeDevelopment = "Development"
annotationKeyRuntimeDevelopmentTarget = "render.crossplane.io/runtime-development-target"
)

// Function returns the Function definition the caller must add to the
Expand All @@ -37,8 +38,11 @@ const (
func (h *Handle) Function() pkgv1.Function {
return pkgv1.Function{
ObjectMeta: metav1.ObjectMeta{
Name: FunctionName,
Annotations: map[string]string{annotationKeyRuntime: annotationValueRuntimeInProcess},
Name: FunctionName,
Annotations: map[string]string{
annotationKeyRuntime: annotationValueRuntimeDevelopment,
annotationKeyRuntimeDevelopmentTarget: h.target,
},
},
}
}
Expand Down
25 changes: 0 additions & 25 deletions cmd/crossplane/render/engine_docker.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ package render
import (
"context"
"os"
"path/filepath"
"runtime"
"strings"

Expand Down Expand Up @@ -146,17 +145,6 @@ func (e *dockerRenderEngine) Render(ctx context.Context, req *renderv1alpha1.Ren
opts = append(opts, docker.RunWithNetworkName(e.network))
}

// Bind-mount the directory of every unix-socket function target into the
// render container at the same path so unix:// targets are reachable.
for _, fn := range getFunctionInputs(req) {
addr := fn.GetAddress()
if !strings.HasPrefix(addr, "unix://") {
continue
}
dir := filepath.Dir(strings.TrimPrefix(addr, "unix://"))
opts = append(opts, docker.RunWithBindMount(dir, dir))
}

e.log.Debug("Running crossplane internal render in Docker", "image", e.image, "network", e.network)

runner := e.runner
Expand Down Expand Up @@ -186,16 +174,3 @@ func (e *dockerRenderEngine) Render(ctx context.Context, req *renderv1alpha1.Ren

return rsp, nil
}

// getFunctionInputs returns the FunctionInput list regardless of which oneof
// variant the RenderRequest carries.
func getFunctionInputs(req *renderv1alpha1.RenderRequest) []*renderv1alpha1.FunctionInput {
switch in := req.GetInput().(type) {
case *renderv1alpha1.RenderRequest_Composite:
return in.Composite.GetFunctions()
case *renderv1alpha1.RenderRequest_Operation:
return in.Operation.GetFunctions()
default:
return nil
}
}
7 changes: 1 addition & 6 deletions cmd/crossplane/render/op/cmd.go
Original file line number Diff line number Diff line change
Expand Up @@ -219,15 +219,10 @@ func (c *Cmd) Run(k *kong.Context, log logging.Logger, sp terminal.SpinnerPrinte
}
defer render.StopFunctionRuntimes(log, fnAddrs)

addrs := fnAddrs.Addresses()
if ctxHandle != nil {
addrs[contextfn.FunctionName] = ctxHandle.Target
}

// Build and execute the render request.
in := render.OperationInputs{
Operation: op,
FunctionAddrs: addrs,
FunctionAddrs: fnAddrs.Addresses(),
RequiredResources: rrs,
RequiredSchemas: rsc,
FunctionCredentials: fcreds,
Expand Down
7 changes: 1 addition & 6 deletions cmd/crossplane/render/xr/cmd.go
Original file line number Diff line number Diff line change
Expand Up @@ -278,16 +278,11 @@ func (c *Cmd) Run(k *kong.Context, log logging.Logger, sp terminal.SpinnerPrinte
}
defer render.StopFunctionRuntimes(log, fnAddrs)

addrs := fnAddrs.Addresses()
if ctxHandle != nil {
addrs[contextfn.FunctionName] = ctxHandle.Target
}

// Build and execute the render request.
in := render.CompositionInputs{
CompositeResource: xr,
Composition: comp,
FunctionAddrs: addrs,
FunctionAddrs: fnAddrs.Addresses(),
ObservedResources: ors,
RequiredResources: rrs,
RequiredSchemas: rsc,
Expand Down
Loading