swirl/biz/docker/task.go

package docker

import (
	"bytes"
	"context"
	"io"
	"math"
	"sort"
	"strconv"

	"github.com/cuigh/swirl/misc"
	"github.com/cuigh/swirl/model"
	"github.com/docker/docker/api/types"
	"github.com/docker/docker/api/types/filters"
	"github.com/docker/docker/api/types/swarm"
	"github.com/docker/docker/client"
	"github.com/docker/docker/pkg/stdcopy"
)

// TaskList return all running tasks of a service or a node.
func TaskList(args *model.TaskListArgs) (infos []*model.TaskInfo, totalCount int, err error) {
	err = mgr.Do(func(ctx context.Context, cli *client.Client) (err error) {
		var (
			tasks []swarm.Task
			opts  = types.TaskListOptions{
				Filters: filters.NewArgs(),
			}
		)

		if args.PageIndex < 1 {
			args.PageIndex = 1
		}
		if args.PageSize < 1 {
			args.PageSize = math.MaxInt32
		}
		if args.Service != "" {
			opts.Filters.Add("service", args.Service)
		}
		if args.Node != "" {
			opts.Filters.Add("node", args.Node)
		}
		if args.Name != "" {
			opts.Filters.Add("name", args.Name)
		}
		if args.State != "" {
			opts.Filters.Add("desired-state", args.State)
		}

		tasks, err = cli.TaskList(ctx, opts)
		totalCount = len(tasks)
		if err == nil && totalCount > 0 {
			sort.Slice(tasks, func(i, j int) bool {
				return tasks[i].UpdatedAt.After(tasks[j].UpdatedAt)
			})
			start, end := misc.Page(totalCount, args.PageIndex, args.PageSize)
			tasks = tasks[start:end]

			nodes := make(map[string]string)
			for _, t := range tasks {
				if _, ok := nodes[t.NodeID]; !ok {
					if n, _, e := cli.NodeInspectWithRaw(ctx, t.NodeID); e == nil {
						if n.Spec.Name == "" {
							nodes[t.NodeID] = n.Description.Hostname
						} else {
							nodes[t.NodeID] = n.Spec.Name
						}
					} else {
						nodes[t.NodeID] = ""
					}
				}
			}

			infos = make([]*model.TaskInfo, len(tasks))
			for i, t := range tasks {
				infos[i] = model.NewTaskInfo(t, nodes[t.NodeID])
			}
		}
		return
	})
	return
}

// TaskInspect return detail information of a task.
func TaskInspect(id string) (task swarm.Task, raw []byte, err error) {
	err = mgr.Do(func(ctx context.Context, cli *client.Client) (err error) {
		task, raw, err = cli.TaskInspectWithRaw(ctx, id)
		return
	})
	return
}

// TaskLogs returns the logs generated by a task in an io.ReadCloser.
// It's up to the caller to close the stream.
func TaskLogs(id string, line int, timestamps bool) (stdout, stderr *bytes.Buffer, err error) {
	var (
		ctx context.Context
		cli *client.Client
		rc  io.ReadCloser
	)

	ctx, cli, err = mgr.Client()
	if err != nil {
		return
	}

	opts := types.ContainerLogsOptions{
		ShowStdout: true,
		ShowStderr: true,
		Tail:       strconv.Itoa(line),
		Timestamps: timestamps,
		//Since: (time.Hour * 24).String()
	}
	if rc, err = cli.TaskLogs(ctx, id, opts); err == nil {
		defer rc.Close()

		stdout = &bytes.Buffer{}
		stderr = &bytes.Buffer{}
		_, err = stdcopy.StdCopy(stdout, stderr, rc)
	}
	return
}