woodpecker/pipeline/backend/docker/docker.go

381 lines
11 KiB
Go

// Copyright 2022 Woodpecker Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package docker
import (
"context"
"io"
"net/http"
"os"
"path/filepath"
"strings"
"github.com/docker/docker/api/types"
"github.com/docker/docker/api/types/network"
"github.com/docker/docker/api/types/volume"
"github.com/docker/go-connections/tlsconfig"
"github.com/moby/moby/client"
"github.com/moby/moby/pkg/jsonmessage"
"github.com/moby/moby/pkg/stdcopy"
"github.com/moby/term"
"github.com/rs/zerolog/log"
"github.com/urfave/cli/v2"
backend "go.woodpecker-ci.org/woodpecker/v2/pipeline/backend/types"
"go.woodpecker-ci.org/woodpecker/v2/shared/utils"
)
type docker struct {
client client.APIClient
enableIPv6 bool
network string
volumes []string
info types.Info
}
const (
networkDriverNAT = "nat"
networkDriverBridge = "bridge"
volumeDriver = "local"
)
// New returns a new Docker Backend.
func New() backend.Backend {
return &docker{
client: nil,
}
}
func (e *docker) Name() string {
return "docker"
}
func (e *docker) IsAvailable(context.Context) bool {
if os.Getenv("DOCKER_HOST") != "" {
return true
}
_, err := os.Stat("/var/run/docker.sock")
return err == nil
}
func httpClientOfOpts(dockerCertPath string, verifyTLS bool) *http.Client {
if dockerCertPath == "" {
return nil
}
options := tlsconfig.Options{
CAFile: filepath.Join(dockerCertPath, "ca.pem"),
CertFile: filepath.Join(dockerCertPath, "cert.pem"),
KeyFile: filepath.Join(dockerCertPath, "key.pem"),
InsecureSkipVerify: !verifyTLS,
}
tlsConf, err := tlsconfig.Client(options)
if err != nil {
log.Error().Err(err).Msg("could not create http client out of docker backend options")
return nil
}
return &http.Client{
Transport: &http.Transport{TLSClientConfig: tlsConf},
CheckRedirect: client.CheckRedirect,
}
}
func (e *docker) Flags() []cli.Flag {
return Flags
}
// Load new client for Docker Backend using environment variables.
func (e *docker) Load(ctx context.Context) (*backend.BackendInfo, error) {
c, ok := ctx.Value(backend.CliContext).(*cli.Context)
if !ok {
return nil, backend.ErrNoCliContextFound
}
var dockerClientOpts []client.Opt
if httpClient := httpClientOfOpts(c.String("backend-docker-cert"), c.Bool("backend-docker-tls-verify")); httpClient != nil {
dockerClientOpts = append(dockerClientOpts, client.WithHTTPClient(httpClient))
}
if dockerHost := c.String("backend-docker-host"); dockerHost != "" {
dockerClientOpts = append(dockerClientOpts, client.WithHost(dockerHost))
}
if dockerAPIVersion := c.String("backend-docker-api-version"); dockerAPIVersion != "" {
dockerClientOpts = append(dockerClientOpts, client.WithVersion(dockerAPIVersion))
} else {
dockerClientOpts = append(dockerClientOpts, client.WithAPIVersionNegotiation())
}
cl, err := client.NewClientWithOpts(dockerClientOpts...)
if err != nil {
return nil, err
}
e.client = cl
e.info, err = cl.Info(ctx)
if err != nil {
return nil, err
}
e.enableIPv6 = c.Bool("backend-docker-ipv6")
e.network = c.String("backend-docker-network")
volumes := strings.Split(c.String("backend-docker-volumes"), ",")
e.volumes = make([]string, 0, len(volumes))
// Validate provided volume definitions
for _, v := range volumes {
if v == "" {
continue
}
parts, err := splitVolumeParts(v)
if err != nil {
log.Error().Err(err).Msgf("invalid volume '%s' provided in WOODPECKER_BACKEND_DOCKER_VOLUMES", v)
continue
}
e.volumes = append(e.volumes, strings.Join(parts, ":"))
}
return &backend.BackendInfo{
Platform: e.info.OSType + "/" + normalizeArchType(e.info.Architecture),
}, nil
}
func (e *docker) SetupWorkflow(ctx context.Context, conf *backend.Config, taskUUID string) error {
log.Trace().Str("taskUUID", taskUUID).Msg("create workflow environment")
for _, vol := range conf.Volumes {
_, err := e.client.VolumeCreate(ctx, volume.CreateOptions{
Name: vol.Name,
Driver: volumeDriver,
})
if err != nil {
return err
}
}
networkDriver := networkDriverBridge
if e.info.OSType == "windows" {
networkDriver = networkDriverNAT
}
for _, n := range conf.Networks {
_, err := e.client.NetworkCreate(ctx, n.Name, types.NetworkCreate{
Driver: networkDriver,
EnableIPv6: e.enableIPv6,
})
if err != nil {
return err
}
}
return nil
}
func (e *docker) StartStep(ctx context.Context, step *backend.Step, taskUUID string) error {
log.Trace().Str("taskUUID", taskUUID).Msgf("start step %s", step.Name)
config := e.toConfig(step)
hostConfig := toHostConfig(step)
containerName := toContainerName(step)
// create pull options with encoded authorization credentials.
pullopts := types.ImagePullOptions{}
if step.AuthConfig.Username != "" && step.AuthConfig.Password != "" {
pullopts.RegistryAuth, _ = encodeAuthToBase64(step.AuthConfig)
}
// automatically pull the latest version of the image if requested
// by the process configuration.
if step.Pull {
responseBody, perr := e.client.ImagePull(ctx, config.Image, pullopts)
if perr == nil {
// TODO(1936): show image pull progress in web-ui
fd, isTerminal := term.GetFdInfo(os.Stdout)
if err := jsonmessage.DisplayJSONMessagesStream(responseBody, os.Stdout, fd, isTerminal, nil); err != nil {
log.Error().Err(err).Msg("DisplayJSONMessagesStream")
}
responseBody.Close()
}
// Fix "Show warning when fail to auth to docker registry"
// (https://web.archive.org/web/20201023145804/https://github.com/drone/drone/issues/1917)
if perr != nil && step.AuthConfig.Password != "" {
return perr
}
}
// add default volumes to the host configuration
hostConfig.Binds = utils.DedupStrings(append(hostConfig.Binds, e.volumes...))
_, err := e.client.ContainerCreate(ctx, config, hostConfig, nil, nil, containerName)
if client.IsErrNotFound(err) {
// automatically pull and try to re-create the image if the
// failure is caused because the image does not exist.
responseBody, perr := e.client.ImagePull(ctx, config.Image, pullopts)
if perr != nil {
return perr
}
// TODO(1936): show image pull progress in web-ui
fd, isTerminal := term.GetFdInfo(os.Stdout)
if err := jsonmessage.DisplayJSONMessagesStream(responseBody, os.Stdout, fd, isTerminal, nil); err != nil {
log.Error().Err(err).Msg("DisplayJSONMessagesStream")
}
responseBody.Close()
_, err = e.client.ContainerCreate(ctx, config, hostConfig, nil, nil, containerName)
}
if err != nil {
return err
}
if len(step.NetworkMode) == 0 {
for _, net := range step.Networks {
err = e.client.NetworkConnect(ctx, net.Name, containerName, &network.EndpointSettings{
Aliases: net.Aliases,
})
if err != nil {
return err
}
}
// join the container to an existing network
if e.network != "" {
err = e.client.NetworkConnect(ctx, e.network, containerName, &network.EndpointSettings{})
if err != nil {
return err
}
}
}
return e.client.ContainerStart(ctx, containerName, startOpts)
}
func (e *docker) WaitStep(ctx context.Context, step *backend.Step, taskUUID string) (*backend.State, error) {
log.Trace().Str("taskUUID", taskUUID).Msgf("wait for step %s", step.Name)
containerName := toContainerName(step)
wait, errc := e.client.ContainerWait(ctx, containerName, "")
select {
case <-wait:
case <-errc:
}
info, err := e.client.ContainerInspect(ctx, containerName)
if err != nil {
return nil, err
}
return &backend.State{
Exited: true,
ExitCode: info.State.ExitCode,
OOMKilled: info.State.OOMKilled,
}, nil
}
func (e *docker) TailStep(ctx context.Context, step *backend.Step, taskUUID string) (io.ReadCloser, error) {
log.Trace().Str("taskUUID", taskUUID).Msgf("tail logs of step %s", step.Name)
logs, err := e.client.ContainerLogs(ctx, toContainerName(step), logsOpts)
if err != nil {
return nil, err
}
rc, wc := io.Pipe()
// de multiplex 'logs' who contains two streams, previously multiplexed together using StdWriter
go func() {
_, _ = stdcopy.StdCopy(wc, wc, logs)
_ = logs.Close()
_ = wc.Close()
}()
return rc, nil
}
func (e *docker) DestroyStep(ctx context.Context, step *backend.Step, taskUUID string) error {
log.Trace().Str("taskUUID", taskUUID).Msgf("stop step %s", step.Name)
containerName := toContainerName(step)
if err := e.client.ContainerKill(ctx, containerName, "9"); err != nil && !isErrContainerNotFoundOrNotRunning(err) {
return err
}
if err := e.client.ContainerRemove(ctx, containerName, removeOpts); err != nil && !isErrContainerNotFoundOrNotRunning(err) {
return err
}
return nil
}
func (e *docker) DestroyWorkflow(ctx context.Context, conf *backend.Config, taskUUID string) error {
log.Trace().Str("taskUUID", taskUUID).Msgf("delete workflow environment")
for _, stage := range conf.Stages {
for _, step := range stage.Steps {
containerName := toContainerName(step)
if err := e.client.ContainerKill(ctx, containerName, "9"); err != nil && !isErrContainerNotFoundOrNotRunning(err) {
log.Error().Err(err).Msgf("could not kill container '%s'", step.Name)
}
if err := e.client.ContainerRemove(ctx, containerName, removeOpts); err != nil && !isErrContainerNotFoundOrNotRunning(err) {
log.Error().Err(err).Msgf("could not remove container '%s'", step.Name)
}
}
}
for _, v := range conf.Volumes {
if err := e.client.VolumeRemove(ctx, v.Name, true); err != nil {
log.Error().Err(err).Msgf("could not remove volume '%s'", v.Name)
}
}
for _, n := range conf.Networks {
if err := e.client.NetworkRemove(ctx, n.Name); err != nil {
log.Error().Err(err).Msgf("could not remove network '%s'", n.Name)
}
}
return nil
}
var (
startOpts = types.ContainerStartOptions{}
removeOpts = types.ContainerRemoveOptions{
RemoveVolumes: true,
RemoveLinks: false,
Force: false,
}
logsOpts = types.ContainerLogsOptions{
Follow: true,
ShowStdout: true,
ShowStderr: true,
Details: false,
Timestamps: false,
}
)
func isErrContainerNotFoundOrNotRunning(err error) bool {
// Error response from daemon: Cannot kill container: ...: No such container: ...
// Error response from daemon: Cannot kill container: ...: Container ... is not running"
// Error response from podman daemon: can only kill running containers. ... is in state exited
// Error: No such container: ...
return err != nil && (strings.Contains(err.Error(), "No such container") || strings.Contains(err.Error(), "is not running") || strings.Contains(err.Error(), "can only kill running containers"))
}
// normalizeArchType converts the arch type reported by docker info into
// the runtime.GOARCH format
// TODO: find out if we we need to convert other arch types too
func normalizeArchType(s string) string {
switch s {
case "x86_64":
return "amd64"
default:
return s
}
}