Tõnis Tiigi
6 years ago
committed by
GitHub
commit
325febc86c
@ -0,0 +1,210 @@
|
||||
package docker
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/docker/docker/api/types"
|
||||
dockertypes "github.com/docker/docker/api/types"
|
||||
"github.com/docker/docker/api/types/container"
|
||||
"github.com/docker/docker/api/types/network"
|
||||
dockerclient "github.com/docker/docker/client"
|
||||
"github.com/docker/docker/pkg/stdcopy"
|
||||
"github.com/moby/buildkit/client"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/tonistiigi/buildx/driver"
|
||||
"github.com/tonistiigi/buildx/util/progress"
|
||||
)
|
||||
|
||||
var buildkitImage = "moby/buildkit:master" // TODO: make this verified and configuratble
|
||||
|
||||
type Driver struct {
|
||||
driver.InitConfig
|
||||
version dockertypes.Version
|
||||
}
|
||||
|
||||
func (d *Driver) Bootstrap(ctx context.Context, l progress.Logger) error {
|
||||
return progress.Wrap("[internal] booting buildkit", l, func(sub progress.SubLogger) error {
|
||||
_, err := d.DockerAPI.ContainerInspect(ctx, d.Name)
|
||||
if err != nil {
|
||||
if dockerclient.IsErrNotFound(err) {
|
||||
return d.create(ctx, sub)
|
||||
}
|
||||
return err
|
||||
}
|
||||
return sub.Wrap("starting container "+d.Name, func() error {
|
||||
if err := d.start(ctx, sub); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := d.wait(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func (d *Driver) create(ctx context.Context, l progress.SubLogger) error {
|
||||
if err := l.Wrap("pulling image "+buildkitImage, func() error {
|
||||
rc, err := d.DockerAPI.ImageCreate(ctx, buildkitImage, types.ImageCreateOptions{})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = io.Copy(ioutil.Discard, rc)
|
||||
return err
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := l.Wrap("creating container "+d.Name, func() error {
|
||||
_, err := d.DockerAPI.ContainerCreate(ctx, &container.Config{
|
||||
Image: buildkitImage,
|
||||
}, &container.HostConfig{
|
||||
Privileged: true,
|
||||
}, &network.NetworkingConfig{}, d.Name)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := d.start(ctx, l); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := d.wait(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *Driver) wait(ctx context.Context) error {
|
||||
try := 0
|
||||
for {
|
||||
if err := d.run(ctx, []string{"buildctl", "debug", "workers"}); err != nil {
|
||||
if try > 10 {
|
||||
return err
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(time.Duration(100+try*20) * time.Millisecond):
|
||||
try++
|
||||
continue
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Driver) exec(ctx context.Context, cmd []string) (string, net.Conn, error) {
|
||||
execConfig := types.ExecConfig{
|
||||
Cmd: cmd,
|
||||
AttachStdin: true,
|
||||
AttachStdout: true,
|
||||
AttachStderr: true,
|
||||
}
|
||||
response, err := d.DockerAPI.ContainerExecCreate(ctx, d.Name, execConfig)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
|
||||
execID := response.ID
|
||||
if execID == "" {
|
||||
return "", nil, errors.New("exec ID empty")
|
||||
}
|
||||
|
||||
resp, err := d.DockerAPI.ContainerExecAttach(ctx, execID, types.ExecStartCheck{})
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
return execID, resp.Conn, nil
|
||||
}
|
||||
|
||||
func (d *Driver) run(ctx context.Context, cmd []string) error {
|
||||
id, conn, err := d.exec(ctx, cmd)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := io.Copy(ioutil.Discard, conn); err != nil {
|
||||
return err
|
||||
}
|
||||
conn.Close()
|
||||
resp, err := d.DockerAPI.ContainerExecInspect(ctx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if resp.ExitCode != 0 {
|
||||
return errors.Errorf("exit code %d", resp.ExitCode)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *Driver) start(ctx context.Context, l progress.SubLogger) error {
|
||||
return d.DockerAPI.ContainerStart(ctx, d.Name, types.ContainerStartOptions{})
|
||||
}
|
||||
|
||||
func (d *Driver) Info(ctx context.Context) (*driver.Info, error) {
|
||||
container, err := d.DockerAPI.ContainerInspect(ctx, d.Name)
|
||||
if err != nil {
|
||||
if dockerclient.IsErrNotFound(err) {
|
||||
return &driver.Info{
|
||||
Status: driver.Terminated,
|
||||
}, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if container.State.Running {
|
||||
return &driver.Info{
|
||||
Status: driver.Running,
|
||||
}, nil
|
||||
}
|
||||
|
||||
return &driver.Info{
|
||||
Status: driver.Stopped,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (d *Driver) Stop(ctx context.Context, force bool) error {
|
||||
return errors.Errorf("stop not implemented for %T", d)
|
||||
}
|
||||
|
||||
func (d *Driver) Rm(ctx context.Context, force bool) error {
|
||||
return errors.Errorf("rm not implemented for %T", d)
|
||||
}
|
||||
|
||||
func (d *Driver) Client(ctx context.Context) (*client.Client, error) {
|
||||
_, conn, err := d.exec(ctx, []string{"buildctl", "dial-stdio"})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
conn = demuxConn(conn)
|
||||
|
||||
return client.New(ctx, "", client.WithDialer(func(string, time.Duration) (net.Conn, error) {
|
||||
return conn, nil
|
||||
}))
|
||||
}
|
||||
|
||||
func demuxConn(c net.Conn) net.Conn {
|
||||
pr, pw := io.Pipe()
|
||||
// TODO: rewrite parser with Reader() to avoid goroutine switch
|
||||
go stdcopy.StdCopy(pw, os.Stdout, c)
|
||||
return &demux{
|
||||
Conn: c,
|
||||
Reader: pr,
|
||||
}
|
||||
}
|
||||
|
||||
type demux struct {
|
||||
net.Conn
|
||||
io.Reader
|
||||
}
|
||||
|
||||
func (d *demux) Read(dt []byte) (int, error) {
|
||||
return d.Reader.Read(dt)
|
||||
}
|
||||
@ -0,0 +1,40 @@
|
||||
package docker
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/pkg/errors"
|
||||
"github.com/tonistiigi/buildx/driver"
|
||||
)
|
||||
|
||||
func init() {
|
||||
driver.Register(&factory{})
|
||||
}
|
||||
|
||||
type factory struct {
|
||||
}
|
||||
|
||||
func (*factory) Name() string {
|
||||
return "docker"
|
||||
}
|
||||
|
||||
func (*factory) Usage() string {
|
||||
return "docker"
|
||||
}
|
||||
|
||||
func (*factory) Priority() int {
|
||||
return 30
|
||||
}
|
||||
|
||||
func (*factory) New(ctx context.Context, cfg driver.InitConfig) (driver.Driver, error) {
|
||||
if cfg.DockerAPI == nil {
|
||||
return nil, errors.Errorf("docker driver requires docker API access")
|
||||
}
|
||||
|
||||
v, err := cfg.DockerAPI.ServerVersion(ctx)
|
||||
if err != nil {
|
||||
return nil, errors.Wrapf(driver.ErrNotConnecting, err.Error())
|
||||
}
|
||||
|
||||
return &Driver{InitConfig: cfg, version: v}, nil
|
||||
}
|
||||
@ -0,0 +1,131 @@
|
||||
package driver
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/moby/buildkit/client"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/tonistiigi/buildx/util/progress"
|
||||
)
|
||||
|
||||
var ErrNotRunning = errors.Errorf("driver not running")
|
||||
var ErrNotConnecting = errors.Errorf("driver not connection")
|
||||
|
||||
type Status int
|
||||
|
||||
const (
|
||||
Terminated Status = iota
|
||||
Starting
|
||||
Running
|
||||
Stopping
|
||||
Stopped
|
||||
)
|
||||
|
||||
type Info struct {
|
||||
Status Status
|
||||
}
|
||||
|
||||
type Driver interface {
|
||||
Bootstrap(context.Context, progress.Logger) error
|
||||
Info(context.Context) (*Info, error)
|
||||
Stop(ctx context.Context, force bool) error
|
||||
Rm(ctx context.Context, force bool) error
|
||||
Client(ctx context.Context) (*client.Client, error)
|
||||
}
|
||||
|
||||
func Boot(ctx context.Context, d Driver, pw progress.Writer) (*client.Client, progress.Writer, error) {
|
||||
try := 0
|
||||
for {
|
||||
info, err := d.Info(ctx)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
try++
|
||||
if info.Status != Running {
|
||||
if try > 2 {
|
||||
return nil, nil, errors.Errorf("failed to bootstrap %T driver in attempts", d)
|
||||
}
|
||||
if err := d.Bootstrap(ctx, func(s *client.SolveStatus) {
|
||||
if pw != nil {
|
||||
pw.Status() <- s
|
||||
}
|
||||
}); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
}
|
||||
|
||||
c, err := d.Client(ctx)
|
||||
if err != nil {
|
||||
if errors.Cause(err) == ErrNotRunning && try <= 2 {
|
||||
continue
|
||||
}
|
||||
return nil, nil, err
|
||||
}
|
||||
return c, newResetWriter(pw), nil
|
||||
}
|
||||
}
|
||||
|
||||
func newResetWriter(in progress.Writer) progress.Writer {
|
||||
w := &pw{Writer: in, status: make(chan *client.SolveStatus), tm: time.Now()}
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case <-in.Done():
|
||||
return
|
||||
case st, ok := <-w.status:
|
||||
if !ok {
|
||||
close(in.Status())
|
||||
return
|
||||
}
|
||||
if w.diff == nil {
|
||||
for _, v := range st.Vertexes {
|
||||
if v.Started != nil {
|
||||
d := v.Started.Sub(w.tm)
|
||||
w.diff = &d
|
||||
}
|
||||
}
|
||||
}
|
||||
if w.diff != nil {
|
||||
for _, v := range st.Vertexes {
|
||||
if v.Started != nil {
|
||||
d := v.Started.Add(-*w.diff)
|
||||
v.Started = &d
|
||||
}
|
||||
if v.Completed != nil {
|
||||
d := v.Completed.Add(-*w.diff)
|
||||
v.Completed = &d
|
||||
}
|
||||
}
|
||||
for _, v := range st.Statuses {
|
||||
if v.Started != nil {
|
||||
d := v.Started.Add(-*w.diff)
|
||||
v.Started = &d
|
||||
}
|
||||
if v.Completed != nil {
|
||||
d := v.Completed.Add(-*w.diff)
|
||||
v.Completed = &d
|
||||
}
|
||||
v.Timestamp = v.Timestamp.Add(-*w.diff)
|
||||
}
|
||||
for _, v := range st.Logs {
|
||||
v.Timestamp = v.Timestamp.Add(-*w.diff)
|
||||
}
|
||||
}
|
||||
in.Status() <- st
|
||||
}
|
||||
}
|
||||
}()
|
||||
return w
|
||||
}
|
||||
|
||||
type pw struct {
|
||||
progress.Writer
|
||||
tm time.Time
|
||||
diff *time.Duration
|
||||
status chan *client.SolveStatus
|
||||
}
|
||||
|
||||
func (p *pw) Status() chan *client.SolveStatus {
|
||||
return p.status
|
||||
}
|
||||
@ -0,0 +1,70 @@
|
||||
package driver
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sort"
|
||||
|
||||
dockerclient "github.com/docker/docker/client"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
type Factory interface {
|
||||
Name() string
|
||||
Usage() string
|
||||
Priority() int // take initConfig?
|
||||
New(ctx context.Context, cfg InitConfig) (Driver, error)
|
||||
}
|
||||
|
||||
type BuildkitConfig struct {
|
||||
// Entitlements []string
|
||||
// Rootless bool
|
||||
}
|
||||
|
||||
type InitConfig struct {
|
||||
// This object needs updates to be generic for different drivers
|
||||
Name string
|
||||
DockerAPI dockerclient.APIClient
|
||||
BuildkitConfig BuildkitConfig
|
||||
Meta map[string]interface{}
|
||||
}
|
||||
|
||||
var drivers map[string]Factory
|
||||
|
||||
func Register(f Factory) {
|
||||
if drivers == nil {
|
||||
drivers = map[string]Factory{}
|
||||
}
|
||||
drivers[f.Name()] = f
|
||||
}
|
||||
|
||||
func GetDefaultFactory() (Factory, error) {
|
||||
if len(drivers) == 0 {
|
||||
return nil, errors.Errorf("no drivers available")
|
||||
}
|
||||
type p struct {
|
||||
f Factory
|
||||
priority int
|
||||
}
|
||||
dd := make([]p, 0, len(drivers))
|
||||
for _, f := range drivers {
|
||||
dd = append(dd, p{f: f, priority: f.Priority()})
|
||||
}
|
||||
sort.Slice(dd, func(i, j int) bool {
|
||||
return dd[i].priority < dd[j].priority
|
||||
})
|
||||
return dd[0].f, nil
|
||||
}
|
||||
|
||||
func GetDriver(ctx context.Context, name string, f Factory, api dockerclient.APIClient) (Driver, error) {
|
||||
if f == nil {
|
||||
var err error
|
||||
f, err = GetDefaultFactory()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return f.New(ctx, InitConfig{
|
||||
Name: name,
|
||||
DockerAPI: api,
|
||||
})
|
||||
}
|
||||
@ -0,0 +1,52 @@
|
||||
package progress
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
|
||||
"github.com/containerd/console"
|
||||
"github.com/moby/buildkit/client"
|
||||
"github.com/moby/buildkit/util/progress/progressui"
|
||||
)
|
||||
|
||||
type printer struct {
|
||||
status chan *client.SolveStatus
|
||||
done <-chan struct{}
|
||||
err error
|
||||
}
|
||||
|
||||
func (p *printer) Done() <-chan struct{} {
|
||||
return p.done
|
||||
}
|
||||
|
||||
func (p *printer) Err() error {
|
||||
return p.err
|
||||
}
|
||||
|
||||
func (p *printer) Status() chan *client.SolveStatus {
|
||||
if p == nil {
|
||||
return nil
|
||||
}
|
||||
return p.status
|
||||
}
|
||||
|
||||
func NewPrinter(ctx context.Context, out *os.File, mode string) Writer {
|
||||
statusCh := make(chan *client.SolveStatus)
|
||||
doneCh := make(chan struct{})
|
||||
|
||||
pw := &printer{
|
||||
status: statusCh,
|
||||
done: doneCh,
|
||||
}
|
||||
|
||||
go func() {
|
||||
var c console.Console
|
||||
if cons, err := console.ConsoleFromFile(out); err == nil && (mode == "auto" || mode == "tty") {
|
||||
c = cons
|
||||
}
|
||||
// not using shared context to not disrupt display but let is finish reporting errors
|
||||
pw.err = progressui.DisplaySolveStatus(ctx, "", c, out, statusCh)
|
||||
close(doneCh)
|
||||
}()
|
||||
return pw
|
||||
}
|
||||
@ -0,0 +1,90 @@
|
||||
package progress
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"github.com/moby/buildkit/client"
|
||||
"github.com/moby/buildkit/identity"
|
||||
"github.com/opencontainers/go-digest"
|
||||
)
|
||||
|
||||
type Logger func(*client.SolveStatus)
|
||||
|
||||
type SubLogger interface {
|
||||
Wrap(name string, fn func() error) error
|
||||
Log(stream int, dt []byte)
|
||||
}
|
||||
|
||||
func Wrap(name string, l Logger, fn func(SubLogger) error) (err error) {
|
||||
dgst := digest.FromBytes([]byte(identity.NewID()))
|
||||
tm := time.Now()
|
||||
l(&client.SolveStatus{
|
||||
Vertexes: []*client.Vertex{{
|
||||
Digest: dgst,
|
||||
Name: name,
|
||||
Started: &tm,
|
||||
}},
|
||||
})
|
||||
|
||||
defer func() {
|
||||
tm2 := time.Now()
|
||||
errMsg := ""
|
||||
if err != nil {
|
||||
errMsg = err.Error()
|
||||
}
|
||||
l(&client.SolveStatus{
|
||||
Vertexes: []*client.Vertex{{
|
||||
Digest: dgst,
|
||||
Name: name,
|
||||
Started: &tm,
|
||||
Completed: &tm2,
|
||||
Error: errMsg,
|
||||
}},
|
||||
})
|
||||
}()
|
||||
|
||||
return fn(&subLogger{dgst, l})
|
||||
}
|
||||
|
||||
type subLogger struct {
|
||||
dgst digest.Digest
|
||||
logger Logger
|
||||
}
|
||||
|
||||
func (sl *subLogger) Wrap(name string, fn func() error) (err error) {
|
||||
tm := time.Now()
|
||||
sl.logger(&client.SolveStatus{
|
||||
Statuses: []*client.VertexStatus{{
|
||||
Vertex: sl.dgst,
|
||||
ID: name,
|
||||
Timestamp: time.Now(),
|
||||
Started: &tm,
|
||||
}},
|
||||
})
|
||||
|
||||
defer func() {
|
||||
tm2 := time.Now()
|
||||
sl.logger(&client.SolveStatus{
|
||||
Statuses: []*client.VertexStatus{{
|
||||
Vertex: sl.dgst,
|
||||
ID: name,
|
||||
Timestamp: time.Now(),
|
||||
Started: &tm,
|
||||
Completed: &tm2,
|
||||
}},
|
||||
})
|
||||
}()
|
||||
|
||||
return fn()
|
||||
}
|
||||
|
||||
func (sl *subLogger) Log(stream int, dt []byte) {
|
||||
sl.logger(&client.SolveStatus{
|
||||
Logs: []*client.VertexLog{{
|
||||
Vertex: sl.dgst,
|
||||
Stream: stream,
|
||||
Data: dt,
|
||||
Timestamp: time.Now(),
|
||||
}},
|
||||
})
|
||||
}
|
||||
@ -0,0 +1,11 @@
|
||||
package progress
|
||||
|
||||
import (
|
||||
"github.com/moby/buildkit/client"
|
||||
)
|
||||
|
||||
type Writer interface {
|
||||
Done() <-chan struct{}
|
||||
Err() error
|
||||
Status() chan *client.SolveStatus
|
||||
}
|
||||
@ -0,0 +1,190 @@
|
||||
package stdcopy // import "github.com/docker/docker/pkg/stdcopy"
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/binary"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// StdType is the type of standard stream
|
||||
// a writer can multiplex to.
|
||||
type StdType byte
|
||||
|
||||
const (
|
||||
// Stdin represents standard input stream type.
|
||||
Stdin StdType = iota
|
||||
// Stdout represents standard output stream type.
|
||||
Stdout
|
||||
// Stderr represents standard error steam type.
|
||||
Stderr
|
||||
// Systemerr represents errors originating from the system that make it
|
||||
// into the multiplexed stream.
|
||||
Systemerr
|
||||
|
||||
stdWriterPrefixLen = 8
|
||||
stdWriterFdIndex = 0
|
||||
stdWriterSizeIndex = 4
|
||||
|
||||
startingBufLen = 32*1024 + stdWriterPrefixLen + 1
|
||||
)
|
||||
|
||||
var bufPool = &sync.Pool{New: func() interface{} { return bytes.NewBuffer(nil) }}
|
||||
|
||||
// stdWriter is wrapper of io.Writer with extra customized info.
|
||||
type stdWriter struct {
|
||||
io.Writer
|
||||
prefix byte
|
||||
}
|
||||
|
||||
// Write sends the buffer to the underneath writer.
|
||||
// It inserts the prefix header before the buffer,
|
||||
// so stdcopy.StdCopy knows where to multiplex the output.
|
||||
// It makes stdWriter to implement io.Writer.
|
||||
func (w *stdWriter) Write(p []byte) (n int, err error) {
|
||||
if w == nil || w.Writer == nil {
|
||||
return 0, errors.New("Writer not instantiated")
|
||||
}
|
||||
if p == nil {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
header := [stdWriterPrefixLen]byte{stdWriterFdIndex: w.prefix}
|
||||
binary.BigEndian.PutUint32(header[stdWriterSizeIndex:], uint32(len(p)))
|
||||
buf := bufPool.Get().(*bytes.Buffer)
|
||||
buf.Write(header[:])
|
||||
buf.Write(p)
|
||||
|
||||
n, err = w.Writer.Write(buf.Bytes())
|
||||
n -= stdWriterPrefixLen
|
||||
if n < 0 {
|
||||
n = 0
|
||||
}
|
||||
|
||||
buf.Reset()
|
||||
bufPool.Put(buf)
|
||||
return
|
||||
}
|
||||
|
||||
// NewStdWriter instantiates a new Writer.
|
||||
// Everything written to it will be encapsulated using a custom format,
|
||||
// and written to the underlying `w` stream.
|
||||
// This allows multiple write streams (e.g. stdout and stderr) to be muxed into a single connection.
|
||||
// `t` indicates the id of the stream to encapsulate.
|
||||
// It can be stdcopy.Stdin, stdcopy.Stdout, stdcopy.Stderr.
|
||||
func NewStdWriter(w io.Writer, t StdType) io.Writer {
|
||||
return &stdWriter{
|
||||
Writer: w,
|
||||
prefix: byte(t),
|
||||
}
|
||||
}
|
||||
|
||||
// StdCopy is a modified version of io.Copy.
|
||||
//
|
||||
// StdCopy will demultiplex `src`, assuming that it contains two streams,
|
||||
// previously multiplexed together using a StdWriter instance.
|
||||
// As it reads from `src`, StdCopy will write to `dstout` and `dsterr`.
|
||||
//
|
||||
// StdCopy will read until it hits EOF on `src`. It will then return a nil error.
|
||||
// In other words: if `err` is non nil, it indicates a real underlying error.
|
||||
//
|
||||
// `written` will hold the total number of bytes written to `dstout` and `dsterr`.
|
||||
func StdCopy(dstout, dsterr io.Writer, src io.Reader) (written int64, err error) {
|
||||
var (
|
||||
buf = make([]byte, startingBufLen)
|
||||
bufLen = len(buf)
|
||||
nr, nw int
|
||||
er, ew error
|
||||
out io.Writer
|
||||
frameSize int
|
||||
)
|
||||
|
||||
for {
|
||||
// Make sure we have at least a full header
|
||||
for nr < stdWriterPrefixLen {
|
||||
var nr2 int
|
||||
nr2, er = src.Read(buf[nr:])
|
||||
nr += nr2
|
||||
if er == io.EOF {
|
||||
if nr < stdWriterPrefixLen {
|
||||
return written, nil
|
||||
}
|
||||
break
|
||||
}
|
||||
if er != nil {
|
||||
return 0, er
|
||||
}
|
||||
}
|
||||
|
||||
stream := StdType(buf[stdWriterFdIndex])
|
||||
// Check the first byte to know where to write
|
||||
switch stream {
|
||||
case Stdin:
|
||||
fallthrough
|
||||
case Stdout:
|
||||
// Write on stdout
|
||||
out = dstout
|
||||
case Stderr:
|
||||
// Write on stderr
|
||||
out = dsterr
|
||||
case Systemerr:
|
||||
// If we're on Systemerr, we won't write anywhere.
|
||||
// NB: if this code changes later, make sure you don't try to write
|
||||
// to outstream if Systemerr is the stream
|
||||
out = nil
|
||||
default:
|
||||
return 0, fmt.Errorf("Unrecognized input header: %d", buf[stdWriterFdIndex])
|
||||
}
|
||||
|
||||
// Retrieve the size of the frame
|
||||
frameSize = int(binary.BigEndian.Uint32(buf[stdWriterSizeIndex : stdWriterSizeIndex+4]))
|
||||
|
||||
// Check if the buffer is big enough to read the frame.
|
||||
// Extend it if necessary.
|
||||
if frameSize+stdWriterPrefixLen > bufLen {
|
||||
buf = append(buf, make([]byte, frameSize+stdWriterPrefixLen-bufLen+1)...)
|
||||
bufLen = len(buf)
|
||||
}
|
||||
|
||||
// While the amount of bytes read is less than the size of the frame + header, we keep reading
|
||||
for nr < frameSize+stdWriterPrefixLen {
|
||||
var nr2 int
|
||||
nr2, er = src.Read(buf[nr:])
|
||||
nr += nr2
|
||||
if er == io.EOF {
|
||||
if nr < frameSize+stdWriterPrefixLen {
|
||||
return written, nil
|
||||
}
|
||||
break
|
||||
}
|
||||
if er != nil {
|
||||
return 0, er
|
||||
}
|
||||
}
|
||||
|
||||
// we might have an error from the source mixed up in our multiplexed
|
||||
// stream. if we do, return it.
|
||||
if stream == Systemerr {
|
||||
return written, fmt.Errorf("error from daemon in stream: %s", string(buf[stdWriterPrefixLen:frameSize+stdWriterPrefixLen]))
|
||||
}
|
||||
|
||||
// Write the retrieved frame (without header)
|
||||
nw, ew = out.Write(buf[stdWriterPrefixLen : frameSize+stdWriterPrefixLen])
|
||||
if ew != nil {
|
||||
return 0, ew
|
||||
}
|
||||
|
||||
// If the frame has not been fully written: error
|
||||
if nw != frameSize {
|
||||
return 0, io.ErrShortWrite
|
||||
}
|
||||
written += int64(nw)
|
||||
|
||||
// Move the rest of the buffer to the beginning
|
||||
copy(buf, buf[frameSize+stdWriterPrefixLen:])
|
||||
// Move the index
|
||||
nr -= frameSize + stdWriterPrefixLen
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,37 @@
|
||||
// Package connhelper provides helpers for connecting to a remote daemon host with custom logic.
|
||||
package connhelper
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"net/url"
|
||||
|
||||
"github.com/docker/cli/cli/connhelper/commandconn"
|
||||
)
|
||||
|
||||
// ConnectionHelper allows to connect to a remote host with custom stream provider binary.
|
||||
type ConnectionHelper struct {
|
||||
// ContextDialer can be passed to grpc.WithContextDialer
|
||||
ContextDialer func(ctx context.Context, addr string) (net.Conn, error)
|
||||
}
|
||||
|
||||
// GetConnectionHelper returns BuildKit-specific connection helper for the given URL.
|
||||
// GetConnectionHelper returns nil without error when no helper is registered for the scheme.
|
||||
//
|
||||
// docker://<container> URL requires BuildKit v0.5.0 or later in the container.
|
||||
func GetConnectionHelper(daemonURL string) (*ConnectionHelper, error) {
|
||||
u, err := url.Parse(daemonURL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
switch scheme := u.Scheme; scheme {
|
||||
case "docker":
|
||||
container := u.Host
|
||||
return &ConnectionHelper{
|
||||
ContextDialer: func(ctx context.Context, addr string) (net.Conn, error) {
|
||||
return commandconn.New(ctx, "docker", "exec", "-i", container, "buildctl", "dial-stdio")
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
Loading…
Reference in New Issue