package docker import ( "context" "encoding/base64" "encoding/json" "errors" "fmt" "os" "strings" "time" "github.com/docker/distribution/reference" "github.com/docker/docker/api/types" "github.com/docker/docker/api/types/container" "github.com/docker/docker/api/types/filters" "github.com/docker/docker/api/types/network" "github.com/docker/docker/api/types/volume" "github.com/docker/docker/client" "github.com/moby/moby/pkg/jsonmessage" "github.com/moby/term" ) // Agent is a Docker client for performing operations that interact // with the Docker engine over REST type Agent struct { authGetter *AuthGetter client *client.Client ctx context.Context label string } // CreateLocalVolumeIfNotExist creates a volume using driver type "local" with the // given name if it does not exist. If the volume does exist but does not contain // the required label (a.label), an error is thrown. func (a *Agent) CreateLocalVolumeIfNotExist(name string) (*types.Volume, error) { volListBody, err := a.client.VolumeList(a.ctx, filters.Args{}) if err != nil { return nil, a.handleDockerClientErr(err, "Could not list volumes") } for _, _vol := range volListBody.Volumes { if contains, ok := _vol.Labels[a.label]; ok && contains == "true" && _vol.Name == name { return _vol, nil } else if !ok && _vol.Name == name { return nil, fmt.Errorf("volume conflict for %s: please remove existing volume and try again", name) } } return a.CreateLocalVolume(name) } // CreateLocalVolume creates a volume using driver type "local" with no // configured options. The equivalent of: // // docker volume create --driver local [name] func (a *Agent) CreateLocalVolume(name string) (*types.Volume, error) { labels := make(map[string]string) labels[a.label] = "true" opts := volume.VolumeCreateBody{ Name: name, Driver: "local", Labels: labels, } vol, err := a.client.VolumeCreate(a.ctx, opts) if err != nil { return nil, a.handleDockerClientErr(err, "Could not create volume "+name) } return &vol, nil } // RemoveLocalVolume removes a volume by name func (a *Agent) RemoveLocalVolume(name string) error { return a.client.VolumeRemove(a.ctx, name, true) } // CreateBridgeNetworkIfNotExist creates a volume using driver type "local" with the // given name if it does not exist. If the volume does exist but does not contain // the required label (a.label), an error is thrown. func (a *Agent) CreateBridgeNetworkIfNotExist(name string) (id string, err error) { networks, err := a.client.NetworkList(a.ctx, types.NetworkListOptions{}) if err != nil { return "", a.handleDockerClientErr(err, "Could not list volumes") } for _, net := range networks { if contains, ok := net.Labels[a.label]; ok && contains == "true" && net.Name == name { return net.ID, nil } else if !ok && net.Name == name { return "", fmt.Errorf("network conflict for %s: please remove existing network and try again", name) } } return a.CreateBridgeNetwork(name) } // CreateBridgeNetwork creates a volume using the default driver type (bridge) // with the CLI label attached func (a *Agent) CreateBridgeNetwork(name string) (id string, err error) { labels := make(map[string]string) labels[a.label] = "true" opts := types.NetworkCreate{ Labels: labels, Attachable: true, } net, err := a.client.NetworkCreate(a.ctx, name, opts) if err != nil { return "", a.handleDockerClientErr(err, "Could not create network "+name) } return net.ID, nil } // ConnectContainerToNetwork attaches a container to a specified network func (a *Agent) ConnectContainerToNetwork(networkID, containerID, containerName string) error { // check if the container is connected already net, err := a.client.NetworkInspect(a.ctx, networkID, types.NetworkInspectOptions{}) if err != nil { return a.handleDockerClientErr(err, "Could not inspect network"+networkID) } for _, cont := range net.Containers { // if container is connected, just return if cont.Name == containerName { return nil } } return a.client.NetworkConnect(a.ctx, networkID, containerID, &network.EndpointSettings{}) } func (a *Agent) TagImage(old, new string) error { return a.client.ImageTag(a.ctx, old, new) } // PullImageEvent represents a response from the Docker API with an image pull event type PullImageEvent struct { Status string `json:"status"` Error string `json:"error"` Progress string `json:"progress"` ProgressDetail struct { Current int `json:"current"` Total int `json:"total"` } `json:"progressDetail"` } var PullImageErrNotFound = fmt.Errorf("Requested image not found") var PullImageErrUnauthorized = fmt.Errorf("Could not pull image: unauthorized") // PullImage pulls an image specified by the image string func (a *Agent) PullImage(image string) error { opts, err := a.getPullOptions(image) if err != nil { return err } // pull the specified image out, err := a.client.ImagePull(a.ctx, image, opts) if err != nil { if client.IsErrNotFound(err) { return PullImageErrNotFound } else if client.IsErrUnauthorized(err) { return PullImageErrUnauthorized } else { return a.handleDockerClientErr(err, "Could not pull image "+image) } } defer out.Close() termFd, isTerm := term.GetFdInfo(os.Stderr) return jsonmessage.DisplayJSONMessagesStream(out, os.Stderr, termFd, isTerm, nil) } // PushImage pushes an image specified by the image string func (a *Agent) PushImage(image string) error { opts, err := a.getPushOptions(image) if err != nil { return err } out, err := a.client.ImagePush( context.Background(), image, opts, ) if out != nil { defer out.Close() } if err != nil { return err } termFd, isTerm := term.GetFdInfo(os.Stderr) return jsonmessage.DisplayJSONMessagesStream(out, os.Stderr, termFd, isTerm, nil) } func (a *Agent) getPullOptions(image string) (types.ImagePullOptions, error) { // check if agent has an auth getter; otherwise, assume public usage if a.authGetter == nil { return types.ImagePullOptions{}, nil } // get using server url serverURL, err := GetServerURLFromTag(image) if err != nil { return types.ImagePullOptions{}, err } user, secret, err := a.authGetter.GetCredentials(serverURL) if err != nil { return types.ImagePullOptions{}, err } var authConfig = types.AuthConfig{ Username: user, Password: secret, ServerAddress: "https://" + serverURL, } authConfigBytes, _ := json.Marshal(authConfig) authConfigEncoded := base64.URLEncoding.EncodeToString(authConfigBytes) return types.ImagePullOptions{ RegistryAuth: authConfigEncoded, Platform: "linux/amd64", }, nil } func (a *Agent) getPushOptions(image string) (types.ImagePushOptions, error) { pullOpts, err := a.getPullOptions(image) return types.ImagePushOptions(pullOpts), err } func GetServerURLFromTag(image string) (string, error) { named, err := reference.ParseNamed(image) if err != nil { return "", err } domain := reference.Domain(named) if domain == "" { // if domain name is empty, use index.docker.io/v1 return "index.docker.io/v1", nil } else if matches := ecrPattern.FindStringSubmatch(image); len(matches) >= 3 { // if this matches ECR, just use the domain name return domain, nil } else if strings.Contains(image, "gcr.io") || strings.Contains(image, "registry.digitalocean.com") { // if this matches GCR or DOCR, use the first path component return fmt.Sprintf("%s/%s", domain, strings.Split(reference.Path(named), "/")[0]), nil } // otherwise, best-guess is to get components of path that aren't the image name pathParts := strings.Split(reference.Path(named), "/") nonImagePath := "" if len(pathParts) > 1 { nonImagePath = strings.Join(pathParts[0:len(pathParts)-1], "/") } if err != nil { return "", err } return fmt.Sprintf("%s/%s", domain, nonImagePath), nil } // WaitForContainerStop waits until a container has stopped to exit func (a *Agent) WaitForContainerStop(id string) error { // wait for container to stop before exit statusCh, errCh := a.client.ContainerWait(a.ctx, id, container.WaitConditionNotRunning) select { case err := <-errCh: if err != nil { return a.handleDockerClientErr(err, "Error waiting for stopped container") } case <-statusCh: } return nil } // WaitForContainerHealthy waits until a container is returning a healthy status. Streak // is the maximum number of failures in a row, while timeout is the length of time between // checks. func (a *Agent) WaitForContainerHealthy(id string, streak int) error { for { cont, err := a.client.ContainerInspect(a.ctx, id) if err != nil { return a.handleDockerClientErr(err, "Error waiting for stopped container") } health := cont.State.Health if health == nil || health.Status == "healthy" { return nil } else if health.FailingStreak >= streak { break } time.Sleep(time.Second) } return errors.New("container not healthy") } // ------------------------- AGENT HELPER FUNCTIONS ------------------------- // func (a *Agent) handleDockerClientErr(err error, errPrefix string) error { if strings.Contains(err.Error(), "Cannot connect to the Docker daemon") { return fmt.Errorf("The Docker daemon must be running in order to start Porter: connection to %s failed", a.client.DaemonHost()) } return fmt.Errorf("%s:%s", errPrefix, err.Error()) }