Merge pull request #1951 from yousong/bugfix/yousong-ansibleserver

ansibleserver: 按步骤记录输出

pkg/ansibleserver/models/ansibleplaybooks.go

@@ -54,11 +54,15 @@ type SAnsiblePlaybook struct {
 	db.SVirtualResourceBase
 
 	Playbook  *ansible.Playbook `nullable:"false" create:"required" get:"user" update:"user"`
-	Output    string            `get:"user"`
+	Output    string            `length:"medium" get:"user"`
 	StartTime time.Time         `list:"user"`
 	EndTime   time.Time         `list:"user"`
 }
 
+const (
+	OutputMaxBytes = 64*1024*1024 - 1
+)
+
 type SAnsiblePlaybookManager struct {
 	db.SVirtualResourceBaseManager
 
@@ -79,6 +83,7 @@ func init() {
 		sessions:    ansible.SessionManager{},
 		sessionsMux: &sync.Mutex{},
 	}
+	AnsiblePlaybookManager.SetVirtualObject(AnsiblePlaybookManager)
 }
 
 func (man *SAnsiblePlaybookManager) ValidateCreateData(ctx context.Context, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, query jsonutils.JSONObject, data *jsonutils.JSONDict) (*jsonutils.JSONDict, error) {
@@ -217,6 +222,7 @@ func (apb *SAnsiblePlaybook) runPlaybook(ctx context.Context, userCred mcclient.
 	} else {
 		pb.PrivateKey = []byte(k)
 	}
+	pb.OutputWriter(&ansiblePlaybookOutputRecorder{apb})
 
 	man.sessions.Add(apb.Id, pb)
 
@@ -250,13 +256,6 @@ func (apb *SAnsiblePlaybook) runPlaybook(ctx context.Context, userCred mcclient.
 				apb.Status = AnsiblePlaybookStatusSucceeded
 			}
 			apb.EndTime = time.Now()
-			// truncate to preserve the tail
-			output := pb.Output()
-			textMax := 64*1024 - 1
-			if len(output) > textMax {
-				output = output[len(output)-textMax:]
-			}
-			apb.Output = string(output)
 			return nil
 		})
 		if err != nil {
@@ -277,3 +276,27 @@ func (apb *SAnsiblePlaybook) stopPlaybook(ctx context.Context, userCred mcclient
 	man.sessions.Stop(apb.Id)
 	return nil
 }
+
+type ansiblePlaybookOutputRecorder struct {
+	apb *SAnsiblePlaybook
+}
+
+func (w *ansiblePlaybookOutputRecorder) Write(p []byte) (n int, err error) {
+	apb := w.apb
+	_, err = db.Update(apb, func() error {
+		cur := apb.Output
+		i := len(p) + len(cur) - OutputMaxBytes
+		if i > 0 {
+			// truncate to preserve the tail
+			apb.Output = cur[:len(cur)-i] + string(p)
+		} else {
+			apb.Output += string(p)
+		}
+		return nil
+	})
+	if err != nil {
+		log.Errorf("ansibleplaybook %s(%s): record output: %v", apb.Name, apb.Id, err)
+		return 0, err
+	}
+	return len(p), nil
+}

pkg/util/ansible/playbook.go

@@ -15,7 +15,6 @@
 package ansible
 
 import (
-	"bytes"
 	"context"
 	"io"
 	"io/ioutil"
@@ -59,7 +58,7 @@ type Playbook struct {
 
 	tmpdir        string
 	noCleanOnExit bool
-	stdio         *bytes.Buffer
+	outputWriter  io.Writer
 	state         pbState
 	stateMux      *sync.Mutex
 }
@@ -68,7 +67,6 @@ func NewPlaybook() *Playbook {
 	pb := &Playbook{
 		state:    pbStateInit,
 		stateMux: &sync.Mutex{},
-		stdio:    &bytes.Buffer{},
 	}
 	return pb
 }
@@ -212,21 +210,11 @@ func (pb *Playbook) Run(ctx context.Context) (err error) {
 			errs = append(errs, errors.WithMessagef(err1, "run module %q, args %q", m.Name, modArgs))
 			return
 		}
-		f := func(r io.Reader) {
-			b := make([]byte, 4096)
-			for {
-				n, err := r.Read(b)
-				if n > 0 {
-					// Mix stdout, stderr
-					pb.stdio.Write(b[:n])
-				}
-				if err != nil {
-					return
-				}
-			}
+		// Mix stdout, stderr
+		if pb.outputWriter != nil {
+			go io.Copy(pb.outputWriter, stdout)
+			go io.Copy(pb.outputWriter, stderr)
 		}
-		go f(stdout)
-		go f(stderr)
 		if err1 := cmd.Wait(); err1 != nil {
 			errs = append(errs, errors.WithMessagef(err1, "wait module %q, args %q", m.Name, modArgs))
 			// continue to next
@@ -235,10 +223,6 @@ func (pb *Playbook) Run(ctx context.Context) (err error) {
 	return nil
 }
 
-// Output returns the stdio output of the playbook
-func (pb *Playbook) Output() []byte {
-	if pb.stdio != nil {
-		return pb.stdio.Bytes()
-	}
-	return nil
+func (pb *Playbook) OutputWriter(w io.Writer) {
+	pb.outputWriter = w
 }

pkg/util/ansible/playbook_test.go

@@ -15,6 +15,7 @@
 package ansible
 
 import (
+	"bytes"
 	"context"
 	"os/exec"
 	"reflect"
@@ -73,8 +74,10 @@ func TestPlaybook(t *testing.T) {
 		}
 	})
 	t.Run("run", func(t *testing.T) {
+		b := &bytes.Buffer{}
+		pb.OutputWriter(b)
 		err := pb.Run(context.TODO())
-		t.Logf("%s", pb.Output())
+		t.Logf("%s", b.String())
 		if err != nil {
 			t.Fatalf("not expecting err: %v", err)
 		}