kraken

1
// Copyright (c) 2016-2019 Uber Technologies, Inc.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
// See the License for the specific language governing permissions and
13
// limitations under the License.
14
package dockerdaemon
15

16
import (
17
	"bytes"
18
	"context"
19
	"fmt"
20
	"io/ioutil"
21
	"net"
22
	"net/http"
23
	"net/url"
24
	"strings"
25
	"syscall"
26
	"time"
27

28
	"golang.org/x/net/context/ctxhttp"
29
)
30

31
const _defaultTimeout = 32 * time.Second
32

33
// DockerClient is a docker daemon client.
34
type DockerClient interface {
35
	PullImage(ctx context.Context, repo, tag string) error
36
}
37

38
type dockerClient struct {
39
	version  string
40
	scheme   string
41
	addr     string
42
	basePath string
43
	registry string
44

45
	client *http.Client
46
}
47

48
// NewDockerClient creates a new DockerClient.
49
func NewDockerClient(config Config, registry string) (DockerClient, error) {
50
	config = config.applyDefaults()
51

52
	client, addr, basePath, err := parseHost(config.DockerHost)
53
	if err != nil {
54
		return nil, fmt.Errorf("parse docker host %q: %s", config.DockerHost, err)
55
	}
56

57
	return &dockerClient{
58
		version:  config.DockerClientVersion,
59
		scheme:   config.DockerScheme,
60
		addr:     addr,
61
		basePath: basePath,
62
		registry: registry,
63
		client:   client,
64
	}, nil
65
}
66

67
// parseHost parses host URL and returns a HTTP client.
68
// This is needed because url.Parse cannot correctly parse url of format
69
// "unix:///...".
70
func parseHost(host string) (*http.Client, string, string, error) {
71
	strs := strings.SplitN(host, "://", 2)
72
	if len(strs) == 1 {
73
		return nil, "", "", fmt.Errorf("unable to parse docker host `%s`", host)
74
	}
75

76
	var basePath string
77
	transport := new(http.Transport)
78

79
	protocol, addr := strs[0], strs[1]
80
	if protocol == "tcp" {
81
		parsed, err := url.Parse("tcp://" + addr)
82
		if err != nil {
83
			return nil, "", "", err
84
		}
85
		addr = parsed.Host
86
		basePath = parsed.Path
87
	} else if protocol == "unix" {
88
		if len(addr) > len(syscall.RawSockaddrUnix{}.Path) {
89
			return nil, "", "", fmt.Errorf("unix socket path %q is too long", addr)
90
		}
91
		transport.DisableCompression = true
92
		transport.Dial = func(_, _ string) (net.Conn, error) {
93
			return net.DialTimeout(protocol, addr, _defaultTimeout)
94
		}
95
	} else {
96
		return nil, "", "", fmt.Errorf("protocol %s not supported", protocol)
97
	}
98

99
	client := &http.Client{
100
		Transport: transport,
101
	}
102
	return client, addr, basePath, nil
103
}
104

105
// ImagePull calls `docker pull` on an image from known registry.
106
func (cli *dockerClient) PullImage(ctx context.Context, repo, tag string) error {
107
	query := url.Values{}
108
	fromImage := fmt.Sprintf("%s/%s", cli.registry, repo)
109
	query.Set("fromImage", fromImage)
110
	query.Set("tag", tag)
111
	headers := map[string][]string{"X-Registry-Auth": {""}}
112
	urlPath := "/images/create"
113

114
	// Construct request. It veries depending on client version.
115
	var apiPath string
116
	if cli.version != "" {
117
		v := strings.TrimPrefix(cli.version, "v")
118
		apiPath = fmt.Sprintf("%s/v%s%s", cli.basePath, v, urlPath)
119
	} else {
120
		apiPath = fmt.Sprintf("%s%s", cli.basePath, urlPath)
121
	}
122
	u := &url.URL{Path: apiPath}
123
	if len(query) > 0 {
124
		u.RawQuery = query.Encode()
125
	}
126
	req, err := http.NewRequest("POST", u.String(), bytes.NewReader([]byte{}))
127
	if err != nil {
128
		return fmt.Errorf("create request: %s", err)
129
	}
130
	req.Header = headers
131
	req.Host = "docker"
132
	req.URL.Host = cli.addr
133
	req.URL.Scheme = cli.scheme
134

135
	resp, err := ctxhttp.Do(ctx, cli.client, req)
136
	if err != nil {
137
		return fmt.Errorf("send post request: %s", err)
138
	}
139
	defer resp.Body.Close()
140
	if resp.StatusCode != 200 {
141
		errMsg, err := ioutil.ReadAll(resp.Body)
142
		if err != nil {
143
			return fmt.Errorf("read error resp: %s", err)
144
		}
145
		return fmt.Errorf("Error posting to %s: code %d, err: %s", urlPath, resp.StatusCode, errMsg)
146
	}
147

148
	// Docker daemon returns 200 early. Close resp.Body after reading all.
149
	if _, err := ioutil.ReadAll(resp.Body); err != nil {
150
		return fmt.Errorf("read resp body: %s", err)
151
	}
152

153
	return nil
154
}
155