package engine import ( "bufio" "bytes" "encoding/json" "fmt" "io" "io/ioutil" "os" "strconv" "strings" "sync" ) // A job is the fundamental unit of work in the docker engine. // Everything docker can do should eventually be exposed as a job. // For example: execute a process in a container, create a new container, // download an archive from the internet, serve the http api, etc. // // The job API is designed after unix processes: a job has a name, arguments, // environment variables, standard streams for input, output and error, and // an exit status which can indicate success (0) or error (anything else). // // One slight variation is that jobs report their status as a string. The // string "0" indicates success, and any other strings indicates an error. // This allows for richer error reporting. // type Job struct { Eng *Engine Name string Args []string env []string Stdin io.Reader Stdout io.Writer Stderr io.Writer handler func(*Job) string status string onExit []func() } // Run executes the job and blocks until the job completes. // If the job returns a failure status, an error is returned // which includes the status. func (job *Job) Run() error { defer func() { var wg sync.WaitGroup for _, f := range job.onExit { wg.Add(1) go func(f func()) { f() wg.Done() }(f) } wg.Wait() }() if job.Stdout != nil && job.Stdout != os.Stdout { job.Stdout = io.MultiWriter(job.Stdout, os.Stdout) } if job.Stderr != nil && job.Stderr != os.Stderr { job.Stderr = io.MultiWriter(job.Stderr, os.Stderr) } job.Eng.Logf("+job %s", job.CallString()) defer func() { job.Eng.Logf("-job %s%s", job.CallString(), job.StatusString()) }() if job.handler == nil { job.status = "command not found" } else { job.status = job.handler(job) } if job.status != "0" { return fmt.Errorf("%s: %s", job.Name, job.status) } return nil } func (job *Job) StdoutParseLines(dst *[]string, limit int) { job.parseLines(job.StdoutPipe(), dst, limit) } func (job *Job) StderrParseLines(dst *[]string, limit int) { job.parseLines(job.StderrPipe(), dst, limit) } func (job *Job) parseLines(src io.Reader, dst *[]string, limit int) { var wg sync.WaitGroup wg.Add(1) go func() { defer wg.Done() scanner := bufio.NewScanner(src) for scanner.Scan() { // If the limit is reached, flush the rest of the source and return if limit > 0 && len(*dst) >= limit { io.Copy(ioutil.Discard, src) return } line := scanner.Text() // Append the line (with delimitor removed) *dst = append(*dst, line) } }() job.onExit = append(job.onExit, wg.Wait) } func (job *Job) StdoutParseString(dst *string) { lines := make([]string, 0, 1) job.StdoutParseLines(&lines, 1) job.onExit = append(job.onExit, func() { if len(lines) >= 1 { *dst = lines[0] } }) } func (job *Job) StderrParseString(dst *string) { lines := make([]string, 0, 1) job.StderrParseLines(&lines, 1) job.onExit = append(job.onExit, func() { *dst = lines[0] }) } func (job *Job) StdoutPipe() io.ReadCloser { r, w := io.Pipe() job.Stdout = w job.onExit = append(job.onExit, func() { w.Close() }) return r } func (job *Job) StderrPipe() io.ReadCloser { r, w := io.Pipe() job.Stderr = w job.onExit = append(job.onExit, func() { w.Close() }) return r } func (job *Job) CallString() string { return fmt.Sprintf("%s(%s)", job.Name, strings.Join(job.Args, ", ")) } func (job *Job) StatusString() string { // FIXME: if a job returns the empty string, it will be printed // as not having returned. // (this only affects String which is a convenience function). if job.status != "" { var okerr string if job.status == "0" { okerr = "OK" } else { okerr = "ERR" } return fmt.Sprintf(" = %s (%s)", okerr, job.status) } return "" } // String returns a human-readable description of `job` func (job *Job) String() string { return fmt.Sprintf("%s.%s%s", job.Eng, job.CallString(), job.StatusString()) } func (job *Job) Getenv(key string) (value string) { for _, kv := range job.env { if strings.Index(kv, "=") == -1 { continue } parts := strings.SplitN(kv, "=", 2) if parts[0] != key { continue } if len(parts) < 2 { value = "" } else { value = parts[1] } } return } func (job *Job) GetenvBool(key string) (value bool) { s := strings.ToLower(strings.Trim(job.Getenv(key), " \t")) if s == "" || s == "0" || s == "no" || s == "false" || s == "none" { return false } return true } func (job *Job) SetenvBool(key string, value bool) { if value { job.Setenv(key, "1") } else { job.Setenv(key, "0") } } func (job *Job) GetenvInt(key string) int64 { s := strings.Trim(job.Getenv(key), " \t") val, err := strconv.ParseInt(s, 10, 64) if err != nil { return -1 } return val } func (job *Job) SetenvInt(key string, value int64) { job.Setenv(key, fmt.Sprintf("%d", value)) } // Returns nil if key not found func (job *Job) GetenvList(key string) []string { sval := job.Getenv(key) if sval == "" { return nil } l := make([]string, 0, 1) if err := json.Unmarshal([]byte(sval), &l); err != nil { l = append(l, sval) } return l } func (job *Job) SetenvJson(key string, value interface{}) error { sval, err := json.Marshal(value) if err != nil { return err } job.Setenv(key, string(sval)) return nil } func (job *Job) SetenvList(key string, value []string) error { return job.SetenvJson(key, value) } func (job *Job) Setenv(key, value string) { job.env = append(job.env, key+"="+value) } // DecodeEnv decodes `src` as a json dictionary, and adds // each decoded key-value pair to the environment. // // If `src` cannot be decoded as a json dictionary, an error // is returned. func (job *Job) DecodeEnv(src io.Reader) error { m := make(map[string]interface{}) if err := json.NewDecoder(src).Decode(&m); err != nil { return err } for k, v := range m { // FIXME: we fix-convert float values to int, because // encoding/json decodes integers to float64, but cannot encode them back. // (See http://golang.org/src/pkg/encoding/json/decode.go#L46) if fval, ok := v.(float64); ok { job.SetenvInt(k, int64(fval)) } else if sval, ok := v.(string); ok { job.Setenv(k, sval) } else if val, err := json.Marshal(v); err == nil { job.Setenv(k, string(val)) } else { job.Setenv(k, fmt.Sprintf("%v", v)) } } return nil } func (job *Job) EncodeEnv(dst io.Writer) error { m := make(map[string]interface{}) for k, v := range job.Environ() { var val interface{} if err := json.Unmarshal([]byte(v), &val); err == nil { // FIXME: we fix-convert float values to int, because // encoding/json decodes integers to float64, but cannot encode them back. // (See http://golang.org/src/pkg/encoding/json/decode.go#L46) if fval, isFloat := val.(float64); isFloat { val = int(fval) } m[k] = val } else { m[k] = v } } if err := json.NewEncoder(dst).Encode(&m); err != nil { return err } return nil } func (job *Job) ExportEnv(dst interface{}) (err error) { defer func() { if err != nil { err = fmt.Errorf("ExportEnv %s", err) } }() var buf bytes.Buffer // step 1: encode/marshal the env to an intermediary json representation if err := job.EncodeEnv(&buf); err != nil { return err } // step 2: decode/unmarshal the intermediary json into the destination object if err := json.NewDecoder(&buf).Decode(dst); err != nil { return err } return nil } func (job *Job) ImportEnv(src interface{}) (err error) { defer func() { if err != nil { err = fmt.Errorf("ImportEnv: %s", err) } }() var buf bytes.Buffer if err := json.NewEncoder(&buf).Encode(src); err != nil { return err } if err := job.DecodeEnv(&buf); err != nil { return err } return nil } func (job *Job) Environ() map[string]string { m := make(map[string]string) for _, kv := range job.env { parts := strings.SplitN(kv, "=", 2) m[parts[0]] = parts[1] } return m } func (job *Job) Logf(format string, args ...interface{}) (n int, err error) { prefixedFormat := fmt.Sprintf("[%s] %s\n", job, strings.TrimRight(format, "\n")) return fmt.Fprintf(job.Stderr, prefixedFormat, args...) } func (job *Job) Printf(format string, args ...interface{}) (n int, err error) { return fmt.Fprintf(job.Stdout, format, args...) } func (job *Job) Errorf(format string, args ...interface{}) (n int, err error) { return fmt.Fprintf(job.Stderr, format, args...) }