job.go 5.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242
  1. package engine
  2. import (
  3. "bytes"
  4. "fmt"
  5. "io"
  6. "strings"
  7. "time"
  8. log "github.com/Sirupsen/logrus"
  9. )
  10. // A job is the fundamental unit of work in the docker engine.
  11. // Everything docker can do should eventually be exposed as a job.
  12. // For example: execute a process in a container, create a new container,
  13. // download an archive from the internet, serve the http api, etc.
  14. //
  15. // The job API is designed after unix processes: a job has a name, arguments,
  16. // environment variables, standard streams for input, output and error, and
  17. // an exit status which can indicate success (0) or error (anything else).
  18. //
  19. // For status, 0 indicates success, and any other integers indicates an error.
  20. // This allows for richer error reporting.
  21. //
  22. type Job struct {
  23. Eng *Engine
  24. Name string
  25. Args []string
  26. env *Env
  27. Stdout *Output
  28. Stderr *Output
  29. Stdin *Input
  30. handler Handler
  31. status Status
  32. end time.Time
  33. closeIO bool
  34. }
  35. type Status int
  36. const (
  37. StatusOK Status = 0
  38. StatusErr Status = 1
  39. StatusNotFound Status = 127
  40. )
  41. // Run executes the job and blocks until the job completes.
  42. // If the job returns a failure status, an error is returned
  43. // which includes the status.
  44. func (job *Job) Run() error {
  45. if job.Eng.IsShutdown() && !job.GetenvBool("overrideShutdown") {
  46. return fmt.Errorf("engine is shutdown")
  47. }
  48. // FIXME: this is a temporary workaround to avoid Engine.Shutdown
  49. // waiting 5 seconds for server/api.ServeApi to complete (which it never will)
  50. // everytime the daemon is cleanly restarted.
  51. // The permanent fix is to implement Job.Stop and Job.OnStop so that
  52. // ServeApi can cooperate and terminate cleanly.
  53. if job.Name != "serveapi" {
  54. job.Eng.l.Lock()
  55. job.Eng.tasks.Add(1)
  56. job.Eng.l.Unlock()
  57. defer job.Eng.tasks.Done()
  58. }
  59. // FIXME: make this thread-safe
  60. // FIXME: implement wait
  61. if !job.end.IsZero() {
  62. return fmt.Errorf("%s: job has already completed", job.Name)
  63. }
  64. // Log beginning and end of the job
  65. if job.Eng.Logging {
  66. log.Infof("+job %s", job.CallString())
  67. defer func() {
  68. log.Infof("-job %s%s", job.CallString(), job.StatusString())
  69. }()
  70. }
  71. var errorMessage = bytes.NewBuffer(nil)
  72. job.Stderr.Add(errorMessage)
  73. if job.handler == nil {
  74. job.Errorf("%s: command not found", job.Name)
  75. job.status = 127
  76. } else {
  77. job.status = job.handler(job)
  78. job.end = time.Now()
  79. }
  80. if job.closeIO {
  81. // Wait for all background tasks to complete
  82. if err := job.Stdout.Close(); err != nil {
  83. return err
  84. }
  85. if err := job.Stderr.Close(); err != nil {
  86. return err
  87. }
  88. if err := job.Stdin.Close(); err != nil {
  89. return err
  90. }
  91. }
  92. if job.status != 0 {
  93. return fmt.Errorf("%s", Tail(errorMessage, 1))
  94. }
  95. return nil
  96. }
  97. func (job *Job) CallString() string {
  98. return fmt.Sprintf("%s(%s)", job.Name, strings.Join(job.Args, ", "))
  99. }
  100. func (job *Job) StatusString() string {
  101. // If the job hasn't completed, status string is empty
  102. if job.end.IsZero() {
  103. return ""
  104. }
  105. var okerr string
  106. if job.status == StatusOK {
  107. okerr = "OK"
  108. } else {
  109. okerr = "ERR"
  110. }
  111. return fmt.Sprintf(" = %s (%d)", okerr, job.status)
  112. }
  113. // String returns a human-readable description of `job`
  114. func (job *Job) String() string {
  115. return fmt.Sprintf("%s.%s%s", job.Eng, job.CallString(), job.StatusString())
  116. }
  117. func (job *Job) Env() *Env {
  118. return job.env
  119. }
  120. func (job *Job) EnvExists(key string) (value bool) {
  121. return job.env.Exists(key)
  122. }
  123. func (job *Job) Getenv(key string) (value string) {
  124. return job.env.Get(key)
  125. }
  126. func (job *Job) GetenvBool(key string) (value bool) {
  127. return job.env.GetBool(key)
  128. }
  129. func (job *Job) SetenvBool(key string, value bool) {
  130. job.env.SetBool(key, value)
  131. }
  132. func (job *Job) GetenvSubEnv(key string) *Env {
  133. return job.env.GetSubEnv(key)
  134. }
  135. func (job *Job) SetenvSubEnv(key string, value *Env) error {
  136. return job.env.SetSubEnv(key, value)
  137. }
  138. func (job *Job) GetenvInt64(key string) int64 {
  139. return job.env.GetInt64(key)
  140. }
  141. func (job *Job) GetenvInt(key string) int {
  142. return job.env.GetInt(key)
  143. }
  144. func (job *Job) SetenvInt64(key string, value int64) {
  145. job.env.SetInt64(key, value)
  146. }
  147. func (job *Job) SetenvInt(key string, value int) {
  148. job.env.SetInt(key, value)
  149. }
  150. // Returns nil if key not found
  151. func (job *Job) GetenvList(key string) []string {
  152. return job.env.GetList(key)
  153. }
  154. func (job *Job) GetenvJson(key string, iface interface{}) error {
  155. return job.env.GetJson(key, iface)
  156. }
  157. func (job *Job) SetenvJson(key string, value interface{}) error {
  158. return job.env.SetJson(key, value)
  159. }
  160. func (job *Job) SetenvList(key string, value []string) error {
  161. return job.env.SetJson(key, value)
  162. }
  163. func (job *Job) Setenv(key, value string) {
  164. job.env.Set(key, value)
  165. }
  166. // DecodeEnv decodes `src` as a json dictionary, and adds
  167. // each decoded key-value pair to the environment.
  168. //
  169. // If `src` cannot be decoded as a json dictionary, an error
  170. // is returned.
  171. func (job *Job) DecodeEnv(src io.Reader) error {
  172. return job.env.Decode(src)
  173. }
  174. func (job *Job) EncodeEnv(dst io.Writer) error {
  175. return job.env.Encode(dst)
  176. }
  177. func (job *Job) ImportEnv(src interface{}) (err error) {
  178. return job.env.Import(src)
  179. }
  180. func (job *Job) Environ() map[string]string {
  181. return job.env.Map()
  182. }
  183. func (job *Job) Logf(format string, args ...interface{}) (n int, err error) {
  184. prefixedFormat := fmt.Sprintf("[%s] %s\n", job, strings.TrimRight(format, "\n"))
  185. return fmt.Fprintf(job.Stderr, prefixedFormat, args...)
  186. }
  187. func (job *Job) Printf(format string, args ...interface{}) (n int, err error) {
  188. return fmt.Fprintf(job.Stdout, format, args...)
  189. }
  190. func (job *Job) Errorf(format string, args ...interface{}) Status {
  191. if format[len(format)-1] != '\n' {
  192. format = format + "\n"
  193. }
  194. fmt.Fprintf(job.Stderr, format, args...)
  195. return StatusErr
  196. }
  197. func (job *Job) Error(err error) Status {
  198. fmt.Fprintf(job.Stderr, "%s\n", err)
  199. return StatusErr
  200. }
  201. func (job *Job) StatusCode() int {
  202. return int(job.status)
  203. }
  204. func (job *Job) SetCloseIO(val bool) {
  205. job.closeIO = val
  206. }