Merge pull request #986 from aluzzardi/socket-support

dagger.#Stream support

cmd/dagger/cmd/input/root.go

@@ -32,6 +32,7 @@ func init() {
 		yamlCmd,
 		listCmd,
 		boolCmd,
+		socketCmd,
 		unsetCmd,
 	)
 }

cmd/dagger/cmd/input/socket.go

@@ -0,0 +1,51 @@
+package input
+
+import (
+	"os"
+
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+	"go.dagger.io/dagger/cmd/dagger/logger"
+	"go.dagger.io/dagger/state"
+)
+
+var socketCmd = &cobra.Command{
+	Use:   "socket <TARGET> <UNIX path>",
+	Short: "Add a socket input",
+	Args:  cobra.ExactArgs(2),
+	PreRun: func(cmd *cobra.Command, args []string) {
+		// Fix Viper bug for duplicate flags:
+		// https://github.com/spf13/viper/issues/233
+		if err := viper.BindPFlags(cmd.Flags()); err != nil {
+			panic(err)
+		}
+	},
+	Run: func(cmd *cobra.Command, args []string) {
+		lg := logger.New()
+		ctx := lg.WithContext(cmd.Context())
+
+		unix := args[1]
+
+		st, err := os.Stat(unix)
+		if err != nil {
+			lg.Fatal().Err(err).Str("path", unix).Msg("invalid unix socket")
+		}
+
+		if st.Mode()&os.ModeSocket == 0 {
+			lg.Fatal().Str("path", unix).Msg("not a unix socket")
+		}
+
+		updateEnvironmentInput(
+			ctx,
+			cmd,
+			args[0],
+			state.SocketInput(unix),
+		)
+	},
+}
+
+func init() {
+	if err := viper.BindPFlags(boolCmd.Flags()); err != nil {
+		panic(err)
+	}
+}

docs/reference/dagger/README.md

@@ -21,3 +21,15 @@ _No input._
 ### dagger.#Secret Outputs
 
 _No output._
+
+## dagger.#Stream
+
+Dagger stream. Can be mounted as a UNIX socket.
+
+### dagger.#Stream Inputs
+
+_No input._
+
+### dagger.#Stream Outputs
+
+_No output._

environment/pipeline.go

@@ -492,6 +492,7 @@ func (p *Pipeline) mount(ctx context.Context, dest string, mnt *compiler.Value)
 			return nil, fmt.Errorf("invalid mount source: %q", s)
 		}
 	}
+
 	// eg. mount: "/foo": secret: mysecret
 	if secret := mnt.Lookup("secret"); secret.Exists() {
 		id, err := getSecretID(secret)
@@ -505,6 +506,28 @@ func (p *Pipeline) mount(ctx context.Context, dest string, mnt *compiler.Value)
 		), nil
 	}
 
+	// eg. mount: "/var/run/docker.sock": stream: mystream
+	if stream := mnt.Lookup("stream"); stream.Exists() {
+		if !stream.HasAttr("stream") {
+			return nil, fmt.Errorf("invalid stream %q: not a stream", stream.Path().String())
+		}
+
+		unixValue := stream.Lookup("unix")
+		if !unixValue.Exists() {
+			return nil, fmt.Errorf("invalid stream %q: not a unix socket", stream.Path().String())
+		}
+
+		unix, err := unixValue.String()
+		if err != nil {
+			return nil, fmt.Errorf("invalid unix path id: %w", err)
+		}
+
+		return llb.AddSSHSocket(
+			llb.SSHID(fmt.Sprintf("unix=%s", unix)),
+			llb.SSHSocketTarget(dest),
+		), nil
+	}
+
 	// eg. mount: "/foo": { from: www.source }
 	if !mnt.Lookup("from").Exists() {
 		return nil, fmt.Errorf("invalid mount: should have %s structure",

solver/socketprovider.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"strings"
 	"time"
 
 	"github.com/moby/buildkit/session"
@@ -13,33 +14,32 @@ import (
 )
 
 const (
-	DockerSocketID   = "docker.sock"
+	unixPrefix = "unix="
-	DockerSocketPath = "/var/run/docker.sock"
 )
 
-type DockerSocketProvider struct {
+type SocketProvider struct {
 }
 
 func NewDockerSocketProvider() session.Attachable {
-	return &DockerSocketProvider{}
+	return &SocketProvider{}
 }
 
-func (sp *DockerSocketProvider) Register(server *grpc.Server) {
+func (sp *SocketProvider) Register(server *grpc.Server) {
 	sshforward.RegisterSSHServer(server, sp)
 }
 
-func (sp *DockerSocketProvider) CheckAgent(ctx context.Context, req *sshforward.CheckAgentRequest) (*sshforward.CheckAgentResponse, error) {
+func (sp *SocketProvider) CheckAgent(ctx context.Context, req *sshforward.CheckAgentRequest) (*sshforward.CheckAgentResponse, error) {
 	id := sshforward.DefaultID
 	if req.ID != "" {
 		id = req.ID
 	}
-	if id != DockerSocketID {
+	if !strings.HasPrefix(id, unixPrefix) {
 		return &sshforward.CheckAgentResponse{}, fmt.Errorf("invalid socket forward key %s", id)
 	}
 	return &sshforward.CheckAgentResponse{}, nil
 }
 
-func (sp *DockerSocketProvider) ForwardAgent(stream sshforward.SSH_ForwardAgentServer) error {
+func (sp *SocketProvider) ForwardAgent(stream sshforward.SSH_ForwardAgentServer) error {
 	id := sshforward.DefaultID
 
 	opts, _ := metadata.FromIncomingContext(stream.Context()) // if no metadata continue with empty object
@@ -48,13 +48,15 @@ func (sp *DockerSocketProvider) ForwardAgent(stream sshforward.SSH_ForwardAgentS
 		id = v[0]
 	}
 
-	if id != DockerSocketID {
+	if !strings.HasPrefix(id, unixPrefix) {
 		return fmt.Errorf("invalid socket forward key %s", id)
 	}
 
-	conn, err := net.DialTimeout("unix", DockerSocketPath, time.Second)
+	id = strings.TrimPrefix(id, unixPrefix)
+
+	conn, err := net.DialTimeout("unix", id, time.Second)
 	if err != nil {
-		return fmt.Errorf("failed to connect to %s: %w", DockerSocketPath, err)
+		return fmt.Errorf("failed to connect to %s: %w", id, err)
 	}
 	defer conn.Close()
 

state/input.go

@@ -37,6 +37,7 @@ type Input struct {
 	YAML   *yamlInput   `yaml:"yaml,omitempty"`
 	File   *fileInput   `yaml:"file,omitempty"`
 	Bool   *boolInput   `yaml:"bool,omitempty"`
+	Socket *socketInput `yaml:"socket,omitempty"`
 }
 
 func (i Input) Compile(key string, state *State) (*compiler.Value, error) {
@@ -59,6 +60,8 @@ func (i Input) Compile(key string, state *State) (*compiler.Value, error) {
 		return i.File.Compile(key, state)
 	case i.Bool != nil:
 		return i.Bool.Compile(key, state)
+	case i.Socket != nil:
+		return i.Socket.Compile(key, state)
 	default:
 		return nil, fmt.Errorf("input has not been set")
 	}
@@ -281,3 +284,22 @@ func (i fileInput) Compile(_ string, _ *State) (*compiler.Value, error) {
 	}
 	return value, nil
 }
+
+// A socket input value
+func SocketInput(data string) Input {
+	i := socketInput{
+		Unix: data,
+	}
+	return Input{
+		Socket: &i,
+	}
+}
+
+type socketInput struct {
+	Unix string `json:"unix,omitempty"`
+}
+
+func (i socketInput) Compile(_ string, _ *State) (*compiler.Value, error) {
+	socketValue := fmt.Sprintf(`{unix: %q}`, i.Unix)
+	return compiler.Compile("", socketValue)
+}

stdlib/dagger/dagger.cue

@@ -14,6 +14,13 @@ import (
 	...
 }
 
+// Dagger stream. Can be mounted as a UNIX socket.
+#Stream: {
+	@dagger(stream)
+
+	unix: string
+}
+
 // Secret value
 #Secret: {
 	@dagger(secret)

tests/compute/dockersocket/main.cue

@@ -1,18 +0,0 @@
-package main
-
-import (
-	"alpha.dagger.io/dagger/op"
-	"alpha.dagger.io/docker"
-)
-
-TestDockerSocket: #up: [
-	op.#Load & {
-		from: docker.#Client
-	},
-
-	op.#Exec & {
-		always: true
-		mount: "/var/run/docker.sock": "docker.sock"
-		args: ["docker", "info"]
-	},
-]

tests/core.bats

@@ -148,9 +148,15 @@ setup() {
     assert_output --partial "secret=mySecret;hash="
 }
 
-@test "compute: docker socket" {
+@test "core: stream" {
-    skip "docker socket support disabled"
+    dagger init
-    run "$DAGGER" compute  "$TESTDIR"/compute/dockersocket
+
+    dagger_new_with_plan test-stream "$TESTDIR"/core/stream
+
+    # Set dir input
+    "$DAGGER" input socket dockersocket /var/run/docker.sock
+
+    "$DAGGER" up
 }
 
 @test "compute: exclude" {

tests/core/stream/main.cue

@@ -0,0 +1,23 @@
+package main
+
+import (
+	"alpha.dagger.io/dagger"
+	"alpha.dagger.io/dagger/op"
+	"alpha.dagger.io/alpine"
+)
+
+dockersocket: dagger.#Stream & dagger.#Input
+
+TestDockerSocket: #up: [
+	op.#Load & {
+		from: alpine.#Image & {
+			package: "docker-cli": true
+		}
+	},
+
+	op.#Exec & {
+		always: true
+		mount: "/var/run/docker.sock": stream: dockersocket
+		args: ["docker", "info"]
+	},
+]