mirror of
https://github.com/fnproject/fn.git
synced 2022-10-28 21:29:17 +03:00
replace heroku registry client w/ docker one
the docker one actually sucks to use, as apparent here. had to add a lot of shit just for it to work, but the heroku registry client has gone stale and is causing issues with our dependencies. didn't expect this to be that hard, but it's over now. good thing we aren't even using this code ^_^. we need to at some point start limiting image sizes and remember this not being fun to figure out to begin with, so anyway, now this works and we don't have to wrestle with different versions of docker in deps (hopefully) anymore.
This commit is contained in:
Reed Allman
@@ -2,46 +2,20 @@ package docker
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"path"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Sirupsen/logrus"
|
||||
manifest "github.com/docker/distribution/manifest/schema1"
|
||||
"github.com/fnproject/fn/api/runner/common"
|
||||
"github.com/fnproject/fn/api/runner/drivers"
|
||||
"github.com/fsouza/go-dockerclient"
|
||||
"github.com/heroku/docker-registry-client/registry"
|
||||
"github.com/opentracing/opentracing-go"
|
||||
)
|
||||
|
||||
const hubURL = "https://registry.hub.docker.com"
|
||||
|
||||
var registryClient = &http.Client{
|
||||
Transport: &http.Transport{
|
||||
Dial: (&net.Dialer{
|
||||
Timeout: 10 * time.Second,
|
||||
KeepAlive: 2 * time.Minute,
|
||||
}).Dial,
|
||||
TLSClientConfig: &tls.Config{
|
||||
ClientSessionCache: tls.NewLRUClientSessionCache(8192),
|
||||
},
|
||||
TLSHandshakeTimeout: 10 * time.Second,
|
||||
MaxIdleConnsPerHost: 32, // TODO tune; we will likely be making lots of requests to same place
|
||||
Proxy: http.ProxyFromEnvironment,
|
||||
IdleConnTimeout: 90 * time.Second,
|
||||
MaxIdleConns: 512,
|
||||
ExpectContinueTimeout: 1 * time.Second,
|
||||
},
|
||||
}
|
||||
|
||||
// A drivers.ContainerTask should implement the Auther interface if it would
|
||||
// like to use not-necessarily-public docker images for any or all task
|
||||
// invocations.
|
||||
@@ -97,122 +71,6 @@ func NewDocker(env *common.Environment, conf drivers.Config) *DockerDriver {
|
||||
}
|
||||
}
|
||||
|
||||
// CheckRegistry will return a sizer, which can be used to check the size of an
|
||||
// image if the returned error is nil. If the error returned is nil, then
|
||||
// authentication against the given credentials was successful, if the
|
||||
// configuration does not specify a config.ServerAddress,
|
||||
// https://hub.docker.com will be tried. CheckRegistry is a package level
|
||||
// method since rkt can also use docker images, we may be interested in using
|
||||
// rkt w/o a docker driver configured; also, we don't have to tote around a
|
||||
// driver in any tasker that may be interested in registry information (2/2
|
||||
// cases thus far).
|
||||
func CheckRegistry(ctx context.Context, image string, config docker.AuthConfiguration) (Sizer, error) {
|
||||
ctx, log := common.LoggerWithFields(ctx, logrus.Fields{"stack": "CheckRegistry"})
|
||||
registry, repo, tag := drivers.ParseImage(image)
|
||||
|
||||
reg, err := registryForConfig(ctx, config, registry)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
mani, err := reg.Manifest(repo, tag)
|
||||
if err != nil {
|
||||
log.WithFields(logrus.Fields{"username": config.Username, "server": config.ServerAddress, "image": image}).WithError(err).Error("Credentials not authorized, trying next.")
|
||||
//if !isAuthError(err) {
|
||||
// // TODO we might retry this, since if this was the registry that was supposed to
|
||||
// // auth the task will erroneously be set to 'error'
|
||||
//}
|
||||
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &sizer{mani, reg, repo}, nil
|
||||
}
|
||||
|
||||
// Sizer returns size information. This interface is liable to contain more
|
||||
// than a size at some point, change as needed.
|
||||
type Sizer interface {
|
||||
Size() (int64, error)
|
||||
}
|
||||
|
||||
type sizer struct {
|
||||
mani *manifest.SignedManifest
|
||||
reg *registry.Registry
|
||||
repo string
|
||||
}
|
||||
|
||||
func (s *sizer) Size() (int64, error) {
|
||||
var sum int64
|
||||
for _, r := range s.mani.References() {
|
||||
desc, err := s.reg.LayerMetadata(s.repo, r.Digest)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
sum += desc.Size
|
||||
}
|
||||
return sum, nil
|
||||
}
|
||||
|
||||
func registryURL(ctx context.Context, addr string) (string, error) {
|
||||
log := common.Logger(ctx)
|
||||
if addr == "" || strings.Contains(addr, "hub.docker.com") || strings.Contains(addr, "index.docker.io") {
|
||||
return hubURL, nil
|
||||
}
|
||||
|
||||
url, err := url.Parse(addr)
|
||||
if err != nil {
|
||||
// TODO we could error the task out from this with a user error but since
|
||||
// we have a list of auths to check, just return the error so as to be
|
||||
// skipped... horrible api as it is
|
||||
log.WithFields(logrus.Fields{"auth_addr": addr}).WithError(err).Error("error parsing server address url, skipping")
|
||||
return "", err
|
||||
}
|
||||
|
||||
if url.Scheme == "" {
|
||||
url.Scheme = "https"
|
||||
}
|
||||
url.Path = strings.TrimSuffix(url.Path, "/")
|
||||
url.Path = strings.TrimPrefix(url.Path, "/v2")
|
||||
url.Path = strings.TrimPrefix(url.Path, "/v1") // just try this, if it fails it fails, not supporting v1
|
||||
return url.String(), nil
|
||||
}
|
||||
|
||||
func isAuthError(err error) bool {
|
||||
// AARGH!
|
||||
if urlError, ok := err.(*url.Error); ok {
|
||||
if httpError, ok := urlError.Err.(*registry.HttpStatusError); ok {
|
||||
if httpError.Response.StatusCode == 401 {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func registryForConfig(ctx context.Context, config docker.AuthConfiguration, reg string) (*registry.Registry, error) {
|
||||
if reg == "" {
|
||||
reg = config.ServerAddress
|
||||
}
|
||||
|
||||
var err error
|
||||
config.ServerAddress, err = registryURL(ctx, reg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Use this instead of registry.New to avoid the Ping().
|
||||
transport := registry.WrapTransport(registryClient.Transport, reg, config.Username, config.Password)
|
||||
r := ®istry.Registry{
|
||||
URL: config.ServerAddress,
|
||||
Client: &http.Client{
|
||||
Transport: transport,
|
||||
},
|
||||
Logf: registry.Quiet,
|
||||
}
|
||||
return r, nil
|
||||
}
|
||||
|
||||
func (drv *DockerDriver) Prepare(ctx context.Context, task drivers.ContainerTask) (drivers.Cookie, error) {
|
||||
ctx, log := common.LoggerWithFields(ctx, logrus.Fields{"stack": "Prepare"})
|
||||
var cmd []string
|
||||
@@ -348,7 +206,7 @@ func (drv *DockerDriver) pullImage(ctx context.Context, task drivers.ContainerTa
|
||||
}
|
||||
|
||||
var err error
|
||||
config.ServerAddress, err = registryURL(ctx, config.ServerAddress)
|
||||
config.ServerAddress, err = registryURL(config.ServerAddress)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
|
||||
"github.com/fnproject/fn/api/runner/common"
|
||||
"github.com/fnproject/fn/api/runner/drivers"
|
||||
"github.com/fsouza/go-dockerclient"
|
||||
)
|
||||
|
||||
type taskDockerTest struct {
|
||||
@@ -89,3 +90,21 @@ func TestRunnerDockerStdin(t *testing.T) {
|
||||
t.Errorf("Test expected output to contain '%s', got '%s'", expect, got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRegistry(t *testing.T) {
|
||||
image := "funcy/hello"
|
||||
|
||||
sizer, err := CheckRegistry(context.Background(), image, docker.AuthConfiguration{})
|
||||
if err != nil {
|
||||
t.Fatal("expected registry check not to fail, got:", err)
|
||||
}
|
||||
|
||||
size, err := sizer.Size()
|
||||
if err != nil {
|
||||
t.Fatal("expected sizer not to fail, got:", err)
|
||||
}
|
||||
|
||||
if size <= 0 {
|
||||
t.Fatalf("expected positive size for image that exists, got size:", size)
|
||||
}
|
||||
}
|
||||
|
||||
207
api/runner/drivers/docker/registry.go
Normal file
207
api/runner/drivers/docker/registry.go
Normal file
@@ -0,0 +1,207 @@
|
||||
package docker
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/docker/distribution"
|
||||
"github.com/docker/distribution/manifest/schema1"
|
||||
"github.com/docker/distribution/manifest/schema2"
|
||||
"github.com/docker/distribution/reference"
|
||||
registry "github.com/docker/distribution/registry/client"
|
||||
"github.com/docker/distribution/registry/client/auth"
|
||||
"github.com/docker/distribution/registry/client/transport"
|
||||
docker "github.com/fsouza/go-dockerclient"
|
||||
"github.com/iron-io/runner/drivers"
|
||||
)
|
||||
|
||||
var (
|
||||
// we need these imported so that they can be unmarshaled properly (yes, docker is mean)
|
||||
_ = schema1.SchemaVersion
|
||||
_ = schema2.SchemaVersion
|
||||
|
||||
registryTransport = &http.Transport{
|
||||
Dial: (&net.Dialer{
|
||||
Timeout: 10 * time.Second,
|
||||
KeepAlive: 2 * time.Minute,
|
||||
}).Dial,
|
||||
TLSClientConfig: &tls.Config{
|
||||
ClientSessionCache: tls.NewLRUClientSessionCache(8192),
|
||||
},
|
||||
TLSHandshakeTimeout: 10 * time.Second,
|
||||
MaxIdleConnsPerHost: 32, // TODO tune; we will likely be making lots of requests to same place
|
||||
Proxy: http.ProxyFromEnvironment,
|
||||
IdleConnTimeout: 90 * time.Second,
|
||||
MaxIdleConns: 512,
|
||||
ExpectContinueTimeout: 1 * time.Second,
|
||||
}
|
||||
)
|
||||
|
||||
const hubURL = "https://registry.hub.docker.com"
|
||||
|
||||
// CheckRegistry will return a sizer, which can be used to check the size of an
|
||||
// image if the returned error is nil. If the error returned is nil, then
|
||||
// authentication against the given credentials was successful, if the
|
||||
// configuration or image do not specify a config.ServerAddress,
|
||||
// https://hub.docker.com will be tried. CheckRegistry is a package level
|
||||
// method since rkt can also use docker images, we may be interested in using
|
||||
// rkt w/o a docker driver configured; also, we don't have to tote around a
|
||||
// driver in any tasker that may be interested in registry information (2/2
|
||||
// cases thus far).
|
||||
func CheckRegistry(ctx context.Context, image string, config docker.AuthConfiguration) (Sizer, error) {
|
||||
regURL, repoName, tag := drivers.ParseImage(image)
|
||||
|
||||
repoNamed, err := reference.WithName(repoName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if regURL == "" {
|
||||
// image address overrides credential address
|
||||
regURL = config.ServerAddress
|
||||
}
|
||||
|
||||
regURL, err = registryURL(regURL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
cm := auth.NewSimpleChallengeManager()
|
||||
|
||||
creds := newCreds(config.Username, config.Password)
|
||||
tran := transport.NewTransport(registryTransport,
|
||||
auth.NewAuthorizer(cm,
|
||||
auth.NewTokenHandler(registryTransport,
|
||||
creds,
|
||||
repoNamed.Name(),
|
||||
"pull",
|
||||
),
|
||||
auth.NewBasicHandler(creds),
|
||||
),
|
||||
)
|
||||
|
||||
tran = &retryWrap{cm, tran}
|
||||
|
||||
repo, err := registry.NewRepository(ctx, repoNamed, regURL, tran)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
manis, err := repo.Manifests(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
mani, err := manis.Get(context.TODO(), "", distribution.WithTag(tag))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
blobs := repo.Blobs(ctx)
|
||||
|
||||
// most registries aren't that great, and won't provide a size for the top
|
||||
// level digest, so we need to sum up all the layers. let this be optional
|
||||
// with the sizer, since tag is good enough to check existence / auth.
|
||||
|
||||
return &sizer{mani, blobs}, nil
|
||||
}
|
||||
|
||||
type retryWrap struct {
|
||||
cm auth.ChallengeManager
|
||||
tran http.RoundTripper
|
||||
}
|
||||
|
||||
func (d *retryWrap) RoundTrip(req *http.Request) (*http.Response, error) {
|
||||
resp, err := d.tran.RoundTrip(req)
|
||||
|
||||
// if it's not authed, we have to add this to the challenge manager,
|
||||
// and then retry it (it will get authed and the challenge then accepted).
|
||||
// why the docker distribution transport doesn't do this for you is
|
||||
// a real testament to what sadists those docker people are.
|
||||
if resp.StatusCode == http.StatusUnauthorized {
|
||||
pingPath := req.URL.Path
|
||||
if v2Root := strings.Index(req.URL.Path, "/v2/"); v2Root != -1 {
|
||||
pingPath = pingPath[:v2Root+4]
|
||||
} else if v1Root := strings.Index(req.URL.Path, "/v1/"); v1Root != -1 {
|
||||
pingPath = pingPath[:v1Root] + "/v2/"
|
||||
}
|
||||
|
||||
// seriously, we have to rewrite this to the ping path,
|
||||
// since looking up challenges strips to this path. YUP. GLHF.
|
||||
ogURL := req.URL.Path
|
||||
resp.Request.URL.Path = pingPath
|
||||
|
||||
d.cm.AddResponse(resp)
|
||||
|
||||
io.Copy(ioutil.Discard, resp.Body)
|
||||
resp.Body.Close()
|
||||
|
||||
// put the original URL path back and try again now...
|
||||
req.URL.Path = ogURL
|
||||
resp, err = d.tran.RoundTrip(req)
|
||||
}
|
||||
return resp, err
|
||||
}
|
||||
|
||||
func newCreds(user, pass string) *creds {
|
||||
return &creds{m: make(map[string]string), user: user, pass: pass}
|
||||
}
|
||||
|
||||
// implement auth.CredentialStore
|
||||
type creds struct {
|
||||
m map[string]string
|
||||
user, pass string
|
||||
}
|
||||
|
||||
func (c *creds) Basic(u *url.URL) (string, string) { return c.user, c.pass }
|
||||
func (c *creds) RefreshToken(u *url.URL, service string) string { return c.m[service] }
|
||||
func (c *creds) SetRefreshToken(u *url.URL, service, token string) { c.m[service] = token }
|
||||
|
||||
// Sizer returns size information. This interface is liable to contain more
|
||||
// than a size at some point, change as needed.
|
||||
type Sizer interface {
|
||||
Size() (int64, error)
|
||||
}
|
||||
|
||||
type sizer struct {
|
||||
mani distribution.Manifest
|
||||
blobs distribution.BlobStore
|
||||
}
|
||||
|
||||
func (s *sizer) Size() (int64, error) {
|
||||
var sum int64
|
||||
for _, r := range s.mani.References() {
|
||||
desc, err := s.blobs.Stat(context.TODO(), r.Digest)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
sum += desc.Size
|
||||
}
|
||||
return sum, nil
|
||||
}
|
||||
|
||||
func registryURL(addr string) (string, error) {
|
||||
if addr == "" || strings.Contains(addr, "hub.docker.com") || strings.Contains(addr, "index.docker.io") {
|
||||
return hubURL, nil
|
||||
}
|
||||
|
||||
uri, err := url.Parse(addr)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
if uri.Scheme == "" {
|
||||
uri.Scheme = "https"
|
||||
}
|
||||
uri.Path = strings.TrimSuffix(uri.Path, "/")
|
||||
uri.Path = strings.TrimPrefix(uri.Path, "/v2")
|
||||
uri.Path = strings.TrimPrefix(uri.Path, "/v1") // just try this, if it fails it fails, not supporting v1
|
||||
return uri.String(), nil
|
||||
}
|
||||
Reference in New Issue
Block a user