Add support for deploy service agent auto updates (#31982) (#33313)

* Add support for ecs agent auto updates

* fix unit test

* Remove unused var

* Addres feedback

* Use list of available AWS database regions

* Run update task on proxy instances

* Revert GenerateAWSOIDCToken

* Move const to start of file

* Address feedback

* Create separate DeployServiceUpdater struct

* Address feedback

- Perform updates in parallel
- Add additional logging
- Add additional documentation

* debug

* Address feedback

- Check OwnershipTags
- Use semaphore pkg
- Release semaphore lease on success

* Make OwnershipTags explicitly required

* Add cluster alert

* Fix typo and update message

* Revert cluster alert

* Update err messages

* Check minimum compatible server version

* Update log msg

api/client/client.go

@@ -3907,6 +3907,22 @@ func (c *Client) ListIntegrations(ctx context.Context, pageSize int, nextKey str
 	return integrations, resp.GetNextKey(), nil
 }
 
+// ListAllIntegrations returns the list of all Integrations.
+func (c *Client) ListAllIntegrations(ctx context.Context) ([]types.Integration, error) {
+	var result []types.Integration
+	var nextKey string
+	for {
+		integrations, nextKey, err := c.ListIntegrations(ctx, 0, nextKey)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		result = append(result, integrations...)
+		if nextKey == "" {
+			return result, nil
+		}
+	}
+}
+
 // GetIntegration returns an Integration by its name.
 func (c *Client) GetIntegration(ctx context.Context, name string) (types.Integration, error) {
 	ig, err := c.integrationsClient().GetIntegration(ctx, &integrationpb.GetIntegrationRequest{

api/types/maintenance.go

@@ -147,6 +147,10 @@ type ClusterMaintenanceConfig interface {
 	// SetAgentUpgradeWindow sets the agent upgrade window.
 	SetAgentUpgradeWindow(win AgentUpgradeWindow)
 
+	// WithinUpgradeWindow returns true if the time is within the configured
+	// upgrade window.
+	WithinUpgradeWindow(t time.Time) bool
+
 	CheckAndSetDefaults() error
 }
 
@@ -229,3 +233,28 @@ func (m *ClusterMaintenanceConfigV1) GetAgentUpgradeWindow() (win AgentUpgradeWi
 func (m *ClusterMaintenanceConfigV1) SetAgentUpgradeWindow(win AgentUpgradeWindow) {
 	m.Spec.AgentUpgrades = &win
 }
+
+// WithinUpgradeWindow returns true if the time is within the configured
+// upgrade window.
+func (m *ClusterMaintenanceConfigV1) WithinUpgradeWindow(t time.Time) bool {
+	upgradeWindow, ok := m.GetAgentUpgradeWindow()
+	if !ok {
+		return false
+	}
+
+	if len(upgradeWindow.Weekdays) == 0 {
+		if int(upgradeWindow.UTCStartHour) == t.Hour() {
+			return true
+		}
+	}
+
+	weekday := t.Weekday().String()
+	for _, upgradeWeekday := range upgradeWindow.Weekdays {
+		if weekday == upgradeWeekday {
+			if int(upgradeWindow.UTCStartHour) == t.Hour() {
+				return true
+			}
+		}
+	}
+	return false
+}

api/types/maintenance_test.go

@@ -214,3 +214,60 @@ func TestWeekdayParser(t *testing.T) {
 		require.Equal(t, tt.expect, day)
 	}
 }
+
+func TestWithinUpgradeWindow(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		desc          string
+		upgradeWindow AgentUpgradeWindow
+		date          string
+		withinWindow  bool
+	}{
+		{
+			desc: "within upgrade window",
+			upgradeWindow: AgentUpgradeWindow{
+				UTCStartHour: 8,
+			},
+			date:         "Mon, 02 Jan 2006 08:04:05 UTC",
+			withinWindow: true,
+		},
+		{
+			desc: "not within upgrade window",
+			upgradeWindow: AgentUpgradeWindow{
+				UTCStartHour: 8,
+			},
+			date:         "Mon, 02 Jan 2006 09:04:05 UTC",
+			withinWindow: false,
+		},
+		{
+			desc: "within upgrade window weekday",
+			upgradeWindow: AgentUpgradeWindow{
+				UTCStartHour: 8,
+				Weekdays:     []string{"Monday"},
+			},
+			date:         "Mon, 02 Jan 2006 08:04:05 UTC",
+			withinWindow: true,
+		},
+		{
+			desc: "not within upgrade window weekday",
+			upgradeWindow: AgentUpgradeWindow{
+				UTCStartHour: 8,
+				Weekdays:     []string{"Tuesday"},
+			},
+			date:         "Mon, 02 Jan 2006 08:04:05 UTC",
+			withinWindow: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.desc, func(t *testing.T) {
+			cmc := NewClusterMaintenanceConfig()
+			cmc.SetAgentUpgradeWindow(tt.upgradeWindow)
+
+			date, err := time.Parse(time.RFC1123, tt.date)
+			require.NoError(t, err)
+			require.Equal(t, tt.withinWindow, cmc.WithinUpgradeWindow(date))
+		})
+	}
+}

lib/authz/permissions.go

@@ -616,6 +616,7 @@ func roleSpecForProxy(clusterName string) types.RoleSpecV6 {
 				types.NewRule(types.KindDatabaseService, services.RO()),
 				types.NewRule(types.KindSAMLIdPServiceProvider, services.RO()),
 				types.NewRule(types.KindUserGroup, services.RO()),
+				types.NewRule(types.KindClusterMaintenanceConfig, services.RO()),
 				types.NewRule(types.KindIntegration, append(services.RO(), types.VerbUse)),
 				// this rule allows cloud proxies to read
 				// plugins of `openai` type, since Assist uses the OpenAI API and runs in Proxy.

lib/automaticupgrades/config.go

@@ -26,6 +26,9 @@ import (
 const (
 	// automaticUpgradesEnvar defines the env var to lookup when deciding whether to enable AutomaticUpgrades feature.
 	automaticUpgradesEnvar = "TELEPORT_AUTOMATIC_UPGRADES"
+
+	// automaticUpgradesChannelEnvar defines a customer automatic upgrades version release channel.
+	automaticUpgradesChannelEnvar = "TELEPORT_AUTOMATIC_UPGRADES_CHANNEL"
 )
 
 // IsEnabled reads the TELEPORT_AUTOMATIC_UPGRADES and returns whether Automatic Upgrades are enabled or disabled.
@@ -46,3 +49,10 @@ func IsEnabled() bool {
 
 	return automaticUpgrades
 }
+
+// GetChannel returns the TELEPORT_AUTOMATIC_UPGRADES_CHANNEL value.
+// Example of an acceptable value for TELEPORT_AUTOMATIC_UPGRADES_CHANNEL is:
+// https://updates.releases.teleport.dev/v1/stable/cloud
+func GetChannel() string {
+	return os.Getenv(automaticUpgradesChannelEnvar)
+}

lib/automaticupgrades/version.go

@@ -34,22 +34,54 @@ const (
 
 	// stableCloudVersionPath is the URL path that returns the current stable/cloud version.
 	stableCloudVersionPath = "/v1/stable/cloud/version"
+
+	// stableCloudCriticalPath is the URL path that returns the stable/cloud critical flag.
+	stableCloudCriticalPath = "/v1/stable/cloud/critical"
 )
 
 // Version returns the version that should be used for installing Teleport Services
 // This is used when installing agents using scripts.
 // Even when Teleport Auth/Proxy is using vX, the agents must always respect this version.
-func Version(ctx context.Context, baseURL string) (string, error) {
-	if baseURL == "" {
-		baseURL = stableCloudVersionBaseURL
+func Version(ctx context.Context, versionURL string) (string, error) {
+	versionURL, err := getVersionURL(versionURL)
+	if err != nil {
+		return "", trace.Wrap(err)
 	}
 
-	fullURL, err := url.JoinPath(baseURL, stableCloudVersionPath)
+	resp, err := sendRequest(ctx, versionURL)
 	if err != nil {
 		return "", trace.Wrap(err)
 	}
 
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, fullURL, nil)
+	return resp, nil
+}
+
+// Critical returns true if a critical upgrade is available.
+func Critical(ctx context.Context, criticalURL string) (bool, error) {
+	criticalURL, err := getCriticalURL(criticalURL)
+	if err != nil {
+		return false, trace.Wrap(err)
+	}
+
+	critical, err := sendRequest(ctx, criticalURL)
+	if err != nil {
+		return false, trace.Wrap(err)
+	}
+
+	// Expects critical endpoint to return either the string "yes" or "no"
+	switch critical {
+	case "yes":
+		return true, nil
+	case "no":
+		return false, nil
+	default:
+		return false, trace.BadParameter("critical endpoint returned an unexpected value: %v", critical)
+	}
+}
+
+// sendRequest sends a GET request to the reqURL and returns the response value
+func sendRequest(ctx context.Context, reqURL string) (string, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil)
 	if err != nil {
 		return "", trace.Wrap(err)
 	}
@@ -69,7 +101,29 @@ func Version(ctx context.Context, baseURL string) (string, error) {
 		return "", trace.BadParameter("invalid status code %d, body: %s", resp.StatusCode, string(body))
 	}
 
-	versionString := strings.TrimSpace(string(body))
+	return strings.TrimSpace(string(body)), trace.Wrap(err)
+}
+
+// getVersionURL returns the versionURL or the default stable/cloud version url.
+func getVersionURL(versionURL string) (string, error) {
+	if versionURL != "" {
+		return versionURL, nil
+	}
+	cloudStableVersionURL, err := url.JoinPath(stableCloudVersionBaseURL, stableCloudVersionPath)
+	if err != nil {
+		return "", trace.Wrap(err)
+	}
+	return cloudStableVersionURL, nil
+}
 
-	return versionString, trace.Wrap(err)
+// getCriticalURL returns the criticalURL or the default stable/cloud critical url.
+func getCriticalURL(criticalURL string) (string, error) {
+	if criticalURL != "" {
+		return criticalURL, nil
+	}
+	cloudStableCriticalURL, err := url.JoinPath(stableCloudVersionBaseURL, stableCloudCriticalPath)
+	if err != nil {
+		return "", trace.Wrap(err)
+	}
+	return cloudStableCriticalURL, nil
 }

lib/automaticupgrades/version_test.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
 	"testing"
 
 	"github.com/gravitational/trace"
@@ -68,19 +69,133 @@ func TestVersion(t *testing.T) {
 	} {
 		t.Run(tt.name, func(t *testing.T) {
 			httpTestServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-				assert.Equal(t, r.URL.Path, "/v1/stable/cloud/version")
+				assert.Equal(t, "/v1/stable/cloud/version", r.URL.Path)
 				w.WriteHeader(tt.mockStatusCode)
 				w.Write([]byte(tt.mockResponseString))
 			}))
 			defer httpTestServer.Close()
 
-			v, err := Version(ctx, httpTestServer.URL)
+			versionURL, err := url.JoinPath(httpTestServer.URL, "/v1/stable/cloud/version")
+			require.NoError(t, err)
+
+			v, err := Version(ctx, versionURL)
 			tt.errCheck(t, err)
 			if err != nil {
 				return
 			}
 
-			require.Equal(t, v, tt.expectedVersion)
+			require.Equal(t, tt.expectedVersion, v)
+		})
+	}
+}
+
+func TestCritical(t *testing.T) {
+	ctx := context.Background()
+
+	isBadParameterErr := func(tt require.TestingT, err error, i ...any) {
+		require.True(tt, trace.IsBadParameter(err), "expected bad parameter, got %v", err)
+	}
+
+	for _, tt := range []struct {
+		name               string
+		mockStatusCode     int
+		mockResponseString string
+		errCheck           require.ErrorAssertionFunc
+		expectedCritical   bool
+	}{
+		{
+			name:               "critical available",
+			mockStatusCode:     http.StatusOK,
+			mockResponseString: "yes\n",
+			errCheck:           require.NoError,
+			expectedCritical:   true,
+		},
+		{
+			name:               "critical is not available",
+			mockStatusCode:     http.StatusOK,
+			mockResponseString: "no\n",
+			errCheck:           require.NoError,
+			expectedCritical:   false,
+		},
+		{
+			name:           "invalid status code (500)",
+			mockStatusCode: http.StatusInternalServerError,
+			errCheck:       isBadParameterErr,
+		},
+		{
+			name:           "invalid status code (403)",
+			mockStatusCode: http.StatusForbidden,
+			errCheck:       isBadParameterErr,
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			httpTestServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				assert.Equal(t, "/v1/stable/cloud/critical", r.URL.Path)
+				w.WriteHeader(tt.mockStatusCode)
+				w.Write([]byte(tt.mockResponseString))
+			}))
+			defer httpTestServer.Close()
+
+			criticalURL, err := url.JoinPath(httpTestServer.URL, "/v1/stable/cloud/critical")
+			require.NoError(t, err)
+
+			v, err := Critical(ctx, criticalURL)
+			tt.errCheck(t, err)
+			if err != nil {
+				return
+			}
+
+			require.Equal(t, tt.expectedCritical, v)
+		})
+	}
+}
+
+func TestGetVersionURL(t *testing.T) {
+	for _, tt := range []struct {
+		name        string
+		versionURL  string
+		expectedURL string
+	}{
+		{
+			name:        "default stable/cloud version url",
+			versionURL:  "",
+			expectedURL: "https://updates.releases.teleport.dev/v1/stable/cloud/version",
+		},
+		{
+			name:        "custom version url",
+			versionURL:  "https://custom.dev/version",
+			expectedURL: "https://custom.dev/version",
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			v, err := getVersionURL(tt.versionURL)
+			require.NoError(t, err)
+			require.Equal(t, tt.expectedURL, v)
+		})
+	}
+}
+
+func TestGetCriticalURL(t *testing.T) {
+	for _, tt := range []struct {
+		name        string
+		criticalURL string
+		expectedURL string
+	}{
+		{
+			name:        "default stable/cloud critical url",
+			criticalURL: "",
+			expectedURL: "https://updates.releases.teleport.dev/v1/stable/cloud/critical",
+		},
+		{
+			name:        "custom critical url",
+			criticalURL: "https://custom.dev/critical",
+			expectedURL: "https://custom.dev/critical",
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			v, err := getCriticalURL(tt.criticalURL)
+			require.NoError(t, err)
+			require.Equal(t, tt.expectedURL, v)
 		})
 	}
 }

lib/cloud/aws/policy_statements.go

@@ -50,7 +50,7 @@ func StatementForECSManageService() *Statement {
 		Actions: []string{
 			"ecs:DescribeClusters", "ecs:CreateCluster", "ecs:PutClusterCapacityProviders",
 			"ecs:DescribeServices", "ecs:CreateService", "ecs:UpdateService",
-			"ecs:RegisterTaskDefinition",
+			"ecs:RegisterTaskDefinition", "ecs:DescribeTaskDefinition", "ecs:DeregisterTaskDefinition",
 
 			// EC2 DescribeSecurityGroups is required so that the user can list the SG and then pick which ones they want to apply to the ECS Service.
 			"ec2:DescribeSecurityGroups",