PMM-12762 Get notified about new version.

managed/cmd/pmm-managed/main.go

@@ -137,7 +137,7 @@ const (
 
 var pprofSemaphore = semaphore.NewWeighted(1)
 
-func addLogsHandler(mux *http.ServeMux, logs *supervisord.Logs) {
+func addLogsHandler(mux *http.ServeMux, logs *server.Logs) {
 	l := logrus.WithField("component", "logs.zip")
 
 	mux.HandleFunc("/logs.zip", func(rw http.ResponseWriter, req *http.Request) {
@@ -147,7 +147,7 @@ func addLogsHandler(mux *http.ServeMux, logs *supervisord.Logs) {
 		if err != nil {
 			l.Debug("Unable to read 'pprof' query param. Using default: pprof=false")
 		}
-		var pprofConfig *supervisord.PprofConfig
+		var pprofConfig *server.PprofConfig
 		if pprofQueryParameter {
 			if !pprofSemaphore.TryAcquire(1) {
 				rw.WriteHeader(http.StatusLocked)
@@ -160,7 +160,7 @@ func addLogsHandler(mux *http.ServeMux, logs *supervisord.Logs) {
 			defer pprofSemaphore.Release(1)
 
 			contextTimeout += pProfProfileDuration + pProfTraceDuration
-			pprofConfig = &supervisord.PprofConfig{
+			pprofConfig = &server.PprofConfig{
 				ProfileDuration: pProfProfileDuration,
 				TraceDuration:   pProfTraceDuration,
 			}
@@ -334,7 +334,7 @@ func runGRPCServer(ctx context.Context, deps *gRPCServerDeps) {
 }
 
 type http1ServerDeps struct {
-	logs       *supervisord.Logs
+	logs       *server.Logs
 	authServer *grafana.AuthServer
 }
 
@@ -875,13 +875,12 @@ func main() { //nolint:cyclop,maintidx
 	connectionCheck := agents.NewConnectionChecker(agentsRegistry)
 	serviceInfoBroker := agents.NewServiceInfoBroker(agentsRegistry)
 
-	pmmUpdateCheck := supervisord.NewPMMUpdateChecker(logrus.WithField("component", "supervisord/pmm-update-checker"))
+	updater := server.NewUpdater(*watchtowerHostF, gRPCMessageMaxSize)
 
-	logs := supervisord.NewLogs(version.FullInfo(), pmmUpdateCheck, vmParams)
+	logs := server.NewLogs(version.FullInfo(), updater, vmParams)
 
 	supervisord := supervisord.New(
 		*supervisordConfigDirF,
-		pmmUpdateCheck,
 		&models.Params{
 			VMParams: vmParams,
 			PGParams: &models.PGParams{
@@ -895,8 +894,7 @@ func main() { //nolint:cyclop,maintidx
 				SSLCertPath: *postgresSSLCertPathF,
 			},
 			HAParams: haParams,
-		},
-		gRPCMessageMaxSize)
+		})
 
 	haService.AddLeaderService(ha.NewStandardService("pmm-agent-runner", func(ctx context.Context) error {
 		return supervisord.StartSupervisedService("pmm-agent")
@@ -971,8 +969,6 @@ func main() { //nolint:cyclop,maintidx
 
 	dumpService := dump.New(db)
 
-	updater := server.NewUpdater(supervisord, *watchtowerHostF)
-
 	serverParams := &server.Params{
 		DB:                   db,
 		VMDB:                 vmdb,

managed/services/server/deps.go

@@ -17,11 +17,11 @@ package server
 
 import (
 	"context"
+	"net/url"
 	"time"
 
 	"github.com/percona/pmm/api/serverpb"
 	"github.com/percona/pmm/managed/models"
-	"github.com/percona/pmm/version"
 )
 
 // healthChecker interface wraps all services that implements the IsReady method to report the
@@ -73,13 +73,13 @@ type vmAlertExternalRules interface {
 // supervisordService is a subset of methods of supervisord.Service used by this package.
 // We use it instead of real type for testing and to avoid dependency cycle.
 type supervisordService interface {
-	InstalledPMMVersion(ctx context.Context) *version.PackageInfo
-	LastCheckUpdatesResult(ctx context.Context) (*version.UpdateCheckResult, time.Time)
-	ForceCheckUpdates(ctx context.Context) error
-
-	StartUpdate() (uint32, error)
-	UpdateRunning() bool
-	UpdateLog(offset uint32) ([]string, uint32, error)
+	//InstalledPMMVersion(ctx context.Context) *version.PackageInfo
+	//LastCheckUpdatesResult(ctx context.Context) (*version.UpdateCheckResult, time.Time)
+	//ForceCheckUpdates(ctx context.Context) error
+	//
+	//StartUpdate() (uint32, error)
+	//UpdateRunning() bool
+	//UpdateLog(offset uint32) ([]string, uint32, error)
 
 	UpdateConfiguration(settings *models.Settings, ssoDetails *models.PerconaSSODetails) error
 }
@@ -108,3 +108,10 @@ type templatesService interface {
 type haService interface {
 	IsLeader() bool
 }
+
+// victoriaMetricsParams is a subset of methods of models.VMParams used by this package.
+// We use it instead of real type to avoid dependency cycle.
+type victoriaMetricsParams interface {
+	ExternalVM() bool
+	URLFor(path string) (*url.URL, error)
+}

managed/services/supervisord/logs.go → managed/services/server/logs.go

@@ -13,7 +13,7 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-package supervisord
+package server
 
 import (
 	"archive/zip"
@@ -56,12 +56,12 @@ type fileContent struct {
 // Logs is responsible for interactions with logs.
 type Logs struct {
 	pmmVersion       string
-	pmmUpdateChecker *PMMUpdateChecker
+	pmmUpdateChecker *Updater
 	vmParams         victoriaMetricsParams
 }
 
 // NewLogs creates a new Logs service.
-func NewLogs(pmmVersion string, pmmUpdateChecker *PMMUpdateChecker, vmParams victoriaMetricsParams) *Logs {
+func NewLogs(pmmVersion string, pmmUpdateChecker *Updater, vmParams victoriaMetricsParams) *Logs {
 	return &Logs{
 		pmmVersion:       pmmVersion,
 		pmmUpdateChecker: pmmUpdateChecker,
@@ -204,7 +204,7 @@ func (l *Logs) files(ctx context.Context, pprofConfig *PprofConfig) []fileConten
 	})
 
 	// update checker installed info
-	b, err = json.Marshal(l.pmmUpdateChecker.Installed(ctx))
+	b, err = json.Marshal(l.pmmUpdateChecker.InstalledPMMVersion())
 	files = append(files, fileContent{
 		Name: "installed.json",
 		Data: b,

managed/services/supervisord/logs_test.go → managed/services/server/logs_test.go

@@ -13,7 +13,7 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-package supervisord
+package server
 
 import (
 	"archive/zip"
@@ -27,7 +27,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -143,10 +142,10 @@ func TestAddAdminSummary(t *testing.T) {
 }
 
 func TestFiles(t *testing.T) {
-	checker := NewPMMUpdateChecker(logrus.WithField("test", t.Name()))
+	updater := &Updater{}
 	params, err := models.NewVictoriaMetricsParams(models.BasePrometheusConfigPath, models.VMBaseURL)
 	require.NoError(t, err)
-	l := NewLogs("2.4.5", checker, params)
+	l := NewLogs("2.4.5", updater, params)
 	ctx := logger.Set(context.Background(), t.Name())
 
 	files := l.files(ctx, nil)
@@ -184,10 +183,10 @@ func TestFiles(t *testing.T) {
 func TestZip(t *testing.T) {
 	t.Skip("FIXME")
 
-	checker := NewPMMUpdateChecker(logrus.WithField("test", t.Name()))
+	updater := &Updater{}
 	params, err := models.NewVictoriaMetricsParams(models.BasePrometheusConfigPath, models.VMBaseURL)
 	require.NoError(t, err)
-	l := NewLogs("2.4.5", checker, params)
+	l := NewLogs("2.4.5", updater, params)
 	ctx := logger.Set(context.Background(), t.Name())
 
 	var buf bytes.Buffer

managed/services/supervisord/pmm_update_checker.go → managed/services/server/pmm_update_checker.go

@@ -13,7 +13,7 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-package supervisord
+package server
 
 import (
 	"bytes"
@@ -24,7 +24,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"golang.org/x/sys/unix"
 
@@ -55,28 +54,9 @@ type PMMUpdateChecker struct {
 
 // NewPMMUpdateChecker returns a PMMUpdateChecker instance that can be shared across different parts of the code.
 // Since this is used inside this package, it could be a singleton, but it would make things mode difficult to test.
-func NewPMMUpdateChecker(l *logrus.Entry) *PMMUpdateChecker {
+func NewPMMUpdateChecker() *PMMUpdateChecker {
 	return &PMMUpdateChecker{
-		l: l,
-	}
-}
-
-// run runs check for updates loop until ctx is canceled.
-func (p *PMMUpdateChecker) run(ctx context.Context) {
-	p.l.Info("Starting...")
-	ticker := time.NewTicker(updateCheckInterval)
-	defer ticker.Stop()
-
-	for {
-		_ = p.check(ctx)
-
-		select {
-		case <-ticker.C:
-			// continue with next loop iteration
-		case <-ctx.Done():
-			p.l.Info("Done.")
-			return
-		}
+		l: logrus.WithField("component", "pmm-update-checker"),
 	}
 }
 
@@ -126,45 +106,3 @@ func (p *PMMUpdateChecker) cmdRun(ctx context.Context, cmdLine string) ([]byte,
 	p.cmdMutex.Unlock()
 	return b, stderr, err
 }
-
-// checkResult returns last `pmm-update -check` result and check time.
-// It may force re-check if last result is empty or too old.
-func (p *PMMUpdateChecker) checkResult(ctx context.Context) (*version.UpdateCheckResult, time.Time) {
-	p.checkRW.RLock()
-	defer p.checkRW.RUnlock()
-
-	if time.Since(p.lastCheckTime) > updateCheckResultFresh {
-		p.checkRW.RUnlock()
-		_ = p.check(ctx)
-		p.checkRW.RLock()
-	}
-
-	return p.lastCheckResult, p.lastCheckTime
-}
-
-// check calls `pmm-update -check` and fills lastInstalledPackageInfo/lastCheckResult/lastCheckTime on success.
-func (p *PMMUpdateChecker) check(ctx context.Context) error {
-	p.checkRW.Lock()
-	defer p.checkRW.Unlock()
-
-	cmdLine := "pmm-update -check"
-	b, stderr, err := p.cmdRun(ctx, cmdLine)
-	if err != nil {
-		p.l.Errorf("%s output: %s. Error: %s", cmdLine, stderr.Bytes(), err)
-		return errors.WithStack(err)
-	}
-
-	var res version.UpdateCheckResult
-	if err = json.Unmarshal(b, &res); err != nil {
-		p.l.Errorf("%s output: %s", cmdLine, stderr.Bytes())
-		return errors.WithStack(err)
-	}
-
-	p.l.Debugf("%s output: %s", cmdLine, stderr.Bytes())
-	p.installedRW.Lock()
-	p.lastInstalledPackageInfo = &res.Installed
-	p.installedRW.Unlock()
-	p.lastCheckResult = &res
-	p.lastCheckTime = time.Now()
-	return nil
-}

managed/services/supervisord/pprof_config.go → managed/services/server/pprof_config.go

@@ -13,7 +13,7 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-package supervisord
+package server
 
 import "time"
 

managed/services/server/server.go

@@ -189,15 +189,14 @@ func (s *Server) Version(ctx context.Context, req *serverpb.VersionRequest) (*se
 		res.Managed.Timestamp = timestamppb.New(t)
 	}
 
-	if v := s.supervisord.InstalledPMMVersion(ctx); v != nil {
-		res.Version = v.Version
-		res.Server = &serverpb.VersionInfo{
-			Version:     v.Version,
-			FullVersion: v.FullVersion,
-		}
-		if v.BuildTime != nil {
-			res.Server.Timestamp = timestamppb.New(*v.BuildTime)
-		}
+	v := s.updater.InstalledPMMVersion()
+	res.Version = v.Version
+	res.Server = &serverpb.VersionInfo{
+		Version:     v.Version,
+		FullVersion: v.FullVersion,
+	}
+	if v.BuildTime != nil {
+		res.Server.Timestamp = timestamppb.New(*v.BuildTime)
 	}
 
 	return res, nil
@@ -298,7 +297,17 @@ func (s *Server) StartUpdate(ctx context.Context, req *serverpb.StartUpdateReque
 		return nil, status.Error(codes.FailedPrecondition, "Updates are disabled via DISABLE_UPDATES environment variable.")
 	}
 
-	err := s.updater.StartUpdate(ctx, req.GetNewImage())
+	newImage := req.GetNewImage()
+	if newImage == "" {
+		latest, err := s.updater.latest(ctx)
+		if err != nil {
+			s.l.WithError(err).Error("Failed to get latest version")
+			newImage = defaultLatestPMMImage
+		} else {
+			newImage = fmt.Sprintf("%s:%s", latest.Repo, latest.Version)
+		}
+	}
+	err := s.updater.StartUpdate(ctx, newImage)
 	if err != nil {
 		return nil, err
 	}
@@ -331,13 +340,13 @@ func (s *Server) UpdateStatus(ctx context.Context, req *serverpb.UpdateStatusReq
 	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
 	defer cancel()
 	for ctx.Err() == nil {
-		done = !s.supervisord.UpdateRunning()
+		done = !s.updater.IsRunning()
 		if done {
 			// give supervisord a second to flush logs to file
 			time.Sleep(time.Second)
 		}
 
-		lines, newOffset, err = s.supervisord.UpdateLog(req.LogOffset)
+		lines, newOffset, err = s.updater.UpdateLog(req.LogOffset)
 		if err != nil {
 			s.l.Warn(err)
 		}