PMM-5680 store logs

agentlocal/agent_local.go

@@ -63,6 +63,11 @@ type Server struct {
 	reloadCloseOnce sync.Once
 }
 
+func (s *Server) mustEmbedUnimplementedAgentLocalServer() {
+	//TODO implement me
+	panic("implement me")
+}
+
 // NewServer creates new server.
 //
 // Caller should call Run.

agents/process/process.go

@@ -19,13 +19,13 @@ package process
 import (
 	"context"
 	"fmt"
+	"github.com/percona/pmm-agent/storelogs"
 	"os/exec"
 	"strings"
 	"time"
 
 	"github.com/percona/pmm/api/inventorypb"
 	"github.com/percona/pmm/utils/pdeathsig"
-	"github.com/sirupsen/logrus"
 	"golang.org/x/sys/unix"
 
 	"github.com/percona/pmm-agent/utils/backoff"
@@ -55,12 +55,11 @@ const (
 // only to avoid stack overflow; there are no extra goroutines for states.
 type Process struct {
 	params  *Params
-	l       *logrus.Entry
 	pl      *processLogger
 	changes chan inventorypb.AgentStatus
 	backoff *backoff.Backoff
 	ctxDone chan struct{}
-
+	logs    *storelogs.LogsStore
 	// recreated on each restart
 	cmd     *exec.Cmd
 	cmdDone chan struct{}
@@ -87,14 +86,14 @@ func (p *Params) String() string {
 }
 
 // New creates new process.
-func New(params *Params, redactWords []string, l *logrus.Entry) *Process {
+func New(params *Params, redactWords []string, sl *storelogs.LogsStore) *Process {
 	return &Process{
 		params:  params,
-		l:       l,
-		pl:      newProcessLogger(l, keepLogLines, redactWords),
+		pl:      newProcessLogger(sl.Entry, keepLogLines, redactWords),
 		changes: make(chan inventorypb.AgentStatus, 10),
 		backoff: backoff.New(backoffMinDelay, backoffMaxDelay),
 		ctxDone: make(chan struct{}),
+		logs:    sl,
 	}
 }
 
@@ -103,14 +102,14 @@ func (p *Process) Run(ctx context.Context) {
 	go p.toStarting()
 
 	<-ctx.Done()
-	p.l.Infof("Process: context canceled.")
+	p.logs.Infof("Process: context canceled.")
 	close(p.ctxDone)
 }
 
 // STARTING -> RUNNING
 // STARTING -> WAITING
 func (p *Process) toStarting() {
-	p.l.Tracef("Process: starting.")
+	p.logs.Tracef("Process: starting.")
 	p.changes <- inventorypb.AgentStatus_STARTING
 
 	p.cmd = exec.Command(p.params.Path, p.params.Args...) //nolint:gosec
@@ -128,7 +127,7 @@ func (p *Process) toStarting() {
 	p.cmdDone = make(chan struct{})
 
 	if err := p.cmd.Start(); err != nil {
-		p.l.Warnf("Process: failed to start: %s.", err)
+		p.logs.Warnf("Process: failed to start: %s.", err)
 		go p.toWaiting()
 		return
 	}
@@ -145,15 +144,15 @@ func (p *Process) toStarting() {
 	case <-t.C:
 		go p.toRunning()
 	case <-p.cmdDone:
-		p.l.Warnf("Process: exited early: %s.", p.cmd.ProcessState)
+		p.logs.Warnf("Process: exited early: %s.", p.cmd.ProcessState)
 		go p.toWaiting()
 	}
 }
 
 // RUNNING -> STOPPING
 // RUNNING -> WAITING
 func (p *Process) toRunning() {
-	p.l.Tracef("Process: running.")
+	p.logs.Tracef("Process: running.")
 	p.changes <- inventorypb.AgentStatus_RUNNING
 
 	p.backoff.Reset()
@@ -162,7 +161,7 @@ func (p *Process) toRunning() {
 	case <-p.ctxDone:
 		go p.toStopping()
 	case <-p.cmdDone:
-		p.l.Warnf("Process: exited: %s.", p.cmd.ProcessState)
+		p.logs.Warnf("Process: exited: %s.", p.cmd.ProcessState)
 		go p.toWaiting()
 	}
 }
@@ -172,7 +171,7 @@ func (p *Process) toRunning() {
 func (p *Process) toWaiting() {
 	delay := p.backoff.Delay()
 
-	p.l.Infof("Process: waiting %s.", delay)
+	p.logs.Infof("Process: waiting %s.", delay)
 	p.changes <- inventorypb.AgentStatus_WAITING
 
 	t := time.NewTimer(delay)
@@ -183,7 +182,7 @@ func (p *Process) toWaiting() {
 		if p.params.Type == inventorypb.AgentType_VM_AGENT {
 			_, err := p.params.TemplateRenderer.RenderFiles(p.params.TemplateParams)
 			if err != nil {
-				p.l.Warnf("Process: failed to regenerate config in %s.", p.params.TemplateRenderer.TempDir)
+				p.logs.Warnf("Process: failed to regenerate config in %s.", p.params.TemplateRenderer.TempDir)
 			}
 		}
 
@@ -195,11 +194,11 @@ func (p *Process) toWaiting() {
 
 // STOPPING -> DONE
 func (p *Process) toStopping() {
-	p.l.Tracef("Process: stopping (sending SIGTERM)...")
+	p.logs.Tracef("Process: stopping (sending SIGTERM)...")
 	p.changes <- inventorypb.AgentStatus_STOPPING
 
 	if err := p.cmd.Process.Signal(unix.SIGTERM); err != nil {
-		p.l.Errorf("Process: failed to send SIGTERM: %s.", err)
+		p.logs.Errorf("Process: failed to send SIGTERM: %s.", err)
 	}
 
 	t := time.NewTimer(killT)
@@ -208,19 +207,19 @@ func (p *Process) toStopping() {
 	case <-p.cmdDone:
 		// nothing
 	case <-t.C:
-		p.l.Warnf("Process: still alive after %s, sending SIGKILL...", killT)
+		p.logs.Warnf("Process: still alive after %s, sending SIGKILL...", killT)
 		if err := p.cmd.Process.Signal(unix.SIGKILL); err != nil {
-			p.l.Errorf("Process: failed to send SIGKILL: %s.", err)
+			p.logs.Errorf("Process: failed to send SIGKILL: %s.", err)
 		}
 		<-p.cmdDone
 	}
 
-	p.l.Infof("Process: exited: %s.", p.cmd.ProcessState)
+	p.logs.Infof("Process: exited: %s.", p.cmd.ProcessState)
 	go p.toDone()
 }
 
 func (p *Process) toDone() {
-	p.l.Trace("Process: done.")
+	p.logs.Trace("Process: done.")
 	p.changes <- inventorypb.AgentStatus_DONE
 
 	close(p.changes)

agents/process/process_child.go

@@ -39,8 +39,9 @@ func main() {
 	logger := logrus.New()
 	logger.SetOutput(ioutil.Discard)
 	l := logrus.NewEntry(logger)
+	ringLogs := ring.New(30)
 
-	p := process.New(&process.Params{Path: "sleep", Args: []string{"100500"}}, nil, l)
+	p := process.New(&process.Params{Path: "sleep", Args: []string{"100500"}}, nil, l, ringLogs)
 	go p.Run(context.Background())
 
 	// Wait until the process is running.

agents/supervisor/supervisor.go

@@ -19,6 +19,7 @@ package supervisor
 import (
 	"context"
 	"fmt"
+	"github.com/percona/pmm-agent/storelogs"
 	"path/filepath"
 	"regexp"
 	"runtime/pprof"
@@ -70,6 +71,7 @@ type agentProcessInfo struct {
 	done           <-chan struct{} // closes when Process.Changes() channel closes
 	requestedState *agentpb.SetStateRequest_AgentProcess
 	listenPort     uint16
+	logs           *storelogs.LogsStore //store logs
 }
 
 // builtinAgentInfo describes built-in Agent.
@@ -119,11 +121,13 @@ func (s *Supervisor) AgentsList() []*agentlocalpb.AgentInfo {
 	res := make([]*agentlocalpb.AgentInfo, 0, len(s.agentProcesses)+len(s.builtinAgents))
 
 	for id, agent := range s.agentProcesses {
+		logs := agent.logs.GetLogs()
 		info := &agentlocalpb.AgentInfo{
 			AgentId:    id,
 			AgentType:  agent.requestedState.Type,
 			Status:     s.lastStatuses[id],
 			ListenPort: uint32(agent.listenPort),
+			Logs:       logs,
 		}
 		res = append(res, info)
 	}
@@ -345,8 +349,9 @@ func (s *Supervisor) startProcess(agentID string, agentProcess *agentpb.SetState
 		"type":      agentType,
 	})
 	l.Debugf("Starting: %s.", processParams)
-
-	process := process.New(processParams, agentProcess.RedactWords, l)
+	ringLog := new(storelogs.LogsStore)
+	ringLog.SetUp(l)
+	process := process.New(processParams, agentProcess.RedactWords, ringLog)
 	go pprof.Do(ctx, pprof.Labels("agentID", agentID, "type", agentType), process.Run)
 
 	done := make(chan struct{})
@@ -368,6 +373,7 @@ func (s *Supervisor) startProcess(agentID string, agentProcess *agentpb.SetState
 		done:           done,
 		requestedState: proto.Clone(agentProcess).(*agentpb.SetStateRequest_AgentProcess),
 		listenPort:     port,
+		logs:           ringLog,
 	}
 	return nil
 }

go.mod

@@ -18,28 +18,28 @@ require (
 	github.com/davecgh/go-spew v1.1.1
 	github.com/go-openapi/runtime v0.19.20
 	github.com/go-sql-driver/mysql v1.5.0
-	github.com/golang/protobuf v1.4.3
-	github.com/grpc-ecosystem/grpc-gateway v1.15.1
+	github.com/golang/protobuf v1.5.2
+	github.com/grpc-ecosystem/grpc-gateway v1.16.0
 	github.com/hashicorp/go-version v1.3.0
 	github.com/lfittl/pg_query_go v1.0.2
 	github.com/lib/pq v1.10.0
 	github.com/percona/exporter_shared v0.7.3
 	github.com/percona/go-mysql v0.0.0-20200630114833-b77f37c0bfa2
 	github.com/percona/percona-toolkit v3.2.1+incompatible
-	github.com/percona/pmm v0.0.0-20220310104949-dd33560bcdde
+	github.com/percona/pmm v0.0.0-20220402125834-c15e36d4238d
 	github.com/pkg/errors v0.9.1
-	github.com/prometheus/client_golang v1.5.1
-	github.com/prometheus/common v0.10.0
+	github.com/prometheus/client_golang v1.12.1
+	github.com/prometheus/common v0.32.1
 	github.com/sirupsen/logrus v1.6.0
 	github.com/stretchr/objx v0.2.0
-	github.com/stretchr/testify v1.6.1
+	github.com/stretchr/testify v1.7.1
 	go.mongodb.org/mongo-driver v1.7.1
-	golang.org/x/sys v0.0.0-20200724161237-0e2f3a69832c
-	google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013
-	google.golang.org/grpc v1.32.0
-	google.golang.org/protobuf v1.25.0
+	golang.org/x/sys v0.0.0-20220114195835-da31bd327af9
+	google.golang.org/genproto v0.0.0-20220302033224-9aa15565e42a
+	google.golang.org/grpc v1.45.0
+	google.golang.org/protobuf v1.27.1
 	gopkg.in/alecthomas/kingpin.v2 v2.2.6
-	gopkg.in/reform.v1 v1.5.0
+	gopkg.in/reform.v1 v1.5.1
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
 )
 
@@ -50,7 +50,7 @@ require (
 	github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d // indirect
 	github.com/asaskevich/govalidator v0.0.0-20200428143746-21a406dcc535 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
-	github.com/cespare/xxhash/v2 v2.1.1 // indirect
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/go-openapi/analysis v0.19.10 // indirect
 	github.com/go-openapi/errors v0.19.6 // indirect
 	github.com/go-openapi/jsonpointer v0.19.3 // indirect
@@ -63,7 +63,6 @@ require (
 	github.com/go-stack/stack v1.8.0 // indirect
 	github.com/gogo/protobuf v1.3.1 // indirect
 	github.com/golang/snappy v0.0.1 // indirect
-	github.com/google/uuid v1.1.2 // indirect
 	github.com/klauspost/compress v1.9.5 // indirect
 	github.com/konsorten/go-windows-terminal-sequences v1.0.3 // indirect
 	github.com/mailru/easyjson v0.7.1 // indirect
@@ -73,15 +72,14 @@ require (
 	github.com/mwitkow/go-proto-validators v0.3.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
-	github.com/prometheus/procfs v0.5.0 // indirect
+	github.com/prometheus/procfs v0.7.3 // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/scram v1.0.2 // indirect
 	github.com/xdg-go/stringprep v1.0.2 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect
 	golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 // indirect
-	golang.org/x/net v0.0.0-20200707034311-ab3426394381 // indirect
+	golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd // indirect
 	golang.org/x/sync v0.0.0-20201207232520-09787c993a3a // indirect
-	golang.org/x/text v0.3.5 // indirect
-	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
-	gopkg.in/yaml.v2 v2.3.0 // indirect
+	golang.org/x/text v0.3.7 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
 )