Enforce a strict upgrade order between stack components

pkg/apis/common/v1/association.go

@@ -67,13 +67,16 @@ type Association interface {
 	SetAssociationStatus(status AssociationStatus)
 }
 
-// AssociationConf holds the association configuration of an Elasticsearch cluster.
+// AssociationConf holds the association configuration of a referenced resource in an association.
 type AssociationConf struct {
 	AuthSecretName string `json:"authSecretName"`
 	AuthSecretKey  string `json:"authSecretKey"`
 	CACertProvided bool   `json:"caCertProvided"`
 	CASecretName   string `json:"caSecretName"`
 	URL            string `json:"url"`
+	// Version of the referenced resource. If a version upgrade is in progress,
+	// matches the lowest running version. May be empty if unknown.
+	Version string `json:"version"`
 }
 
 // IsConfigured returns true if all the fields are set.
@@ -139,3 +142,10 @@ func (ac *AssociationConf) GetURL() string {
 	}
 	return ac.URL
 }
+
+func (ac *AssociationConf) GetVersion() string {
+	if ac == nil {
+		return ""
+	}
+	return ac.Version
+}

pkg/controller/apmserver/controller.go

@@ -25,6 +25,7 @@ import (
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/operator"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/reconciler"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/tracing"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/common/version"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/watches"
 	"github.com/elastic/cloud-on-k8s/pkg/utils/k8s"
 
@@ -251,6 +252,15 @@ func (r *ReconcileApmServer) doReconcile(ctx context.Context, request reconcile.
 		return res, err
 	}
 
+	asVersion, err := version.Parse(as.Spec.Version)
+	if err != nil {
+		return reconcile.Result{}, err
+	}
+	logger := log.WithValues("namespace", as.Namespace, "as_name", as.Name)
+	if !association.AllowVersion(*asVersion, as, logger) {
+		return reconcile.Result{}, nil // will eventually retry
+	}
+
 	state, err = r.reconcileApmServerDeployment(ctx, state, as)
 	if err != nil {
 		if apierrors.IsConflict(err) {

pkg/controller/association/conf.go

@@ -10,6 +10,7 @@ import (
 	"reflect"
 	"unsafe"
 
+	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
 	"go.elastic.co/apm"
 	v1 "k8s.io/api/core/v1"
@@ -21,6 +22,7 @@ import (
 
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/events"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/tracing"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/common/version"
 
 	commonv1 "github.com/elastic/cloud-on-k8s/pkg/apis/common/v1"
 	"github.com/elastic/cloud-on-k8s/pkg/utils/k8s"
@@ -103,6 +105,38 @@ func ElasticsearchAuthSettings(c k8s.Client, association commonv1.Association) (
 	return assocConf.AuthSecretKey, string(data), nil
 }
 
+// AllowVersion returns true if the given resourceVersion is lower or equal to the associations version.
+// For example: Kibana in version 7.8.0 cannot be deployed if its Elasticsearch association reports version 7.7.0.
+// Referenced resources version is parsed from the association conf annotation.
+func AllowVersion(resourceVersion version.Version, associated commonv1.Associated, logger logr.Logger) bool {
+	for _, assoc := range associated.GetAssociations() {
+		assocRef := assoc.AssociationRef()
+		if !assocRef.IsDefined() {
+			// no association specified, move on
+			continue
+		}
+		if assoc.AssociationConf() == nil || assoc.AssociationConf().Version == "" {
+			// no conf reported yet, this may be the initial resource creation
+			logger.Info("Delaying version deployment since the version of an associated resource is not reported yet",
+				"version", resourceVersion, "ref_namespace", assocRef.Namespace, "ref_name", assocRef.Name)
+			return false
+		}
+		refVer, err := version.Parse(assoc.AssociationConf().Version)
+		if err != nil {
+			logger.Error(err, "Invalid version found in association conf", "association_version", refVer)
+			return false
+		}
+		if !refVer.IsSameOrAfter(resourceVersion) {
+			// the version of the referenced resource (example: Elasticsearch) is lower than
+			// the desired version of the reconciled resource (example: Kibana)
+			logger.Info("Delaying version deployment since an associated resource is not upgraded yet",
+				"version", resourceVersion, "ref_namespace", assocRef.Namespace, "ref_name", assocRef.Name, "ref_version", refVer)
+			return false
+		}
+	}
+	return true
+}
+
 // GetAssociationConf extracts the association configuration from the given object by reading the annotations.
 func GetAssociationConf(association commonv1.Association) (*commonv1.AssociationConf, error) {
 	accessor := meta.NewAccessor()

pkg/controller/association/conf_test.go

@@ -18,6 +18,7 @@ import (
 	apmv1 "github.com/elastic/cloud-on-k8s/pkg/apis/apm/v1"
 	commonv1 "github.com/elastic/cloud-on-k8s/pkg/apis/common/v1"
 	kbv1 "github.com/elastic/cloud-on-k8s/pkg/apis/kibana/v1"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/common/version"
 	"github.com/elastic/cloud-on-k8s/pkg/utils/k8s"
 )
 
@@ -411,3 +412,90 @@ func TestRemoveAssociationConf(t *testing.T) {
 	require.EqualValues(t, 1, got.Spec.Count)
 	require.Nil(t, got.AssociationConf())
 }
+
+func TestAllowVersion(t *testing.T) {
+	apmNoAssoc := &apmv1.ApmServer{}
+	apmTwoAssoc := &apmv1.ApmServer{Spec: apmv1.ApmServerSpec{
+		ElasticsearchRef: commonv1.ObjectSelector{Name: "some-es"}, KibanaRef: commonv1.ObjectSelector{Name: "some-kb"}}}
+	apmTwoAssocWithVersions := func(versions []string) *apmv1.ApmServer {
+		apm := apmTwoAssoc.DeepCopy()
+		for i, assoc := range apm.GetAssociations() {
+			assoc.SetAssociationConf(&commonv1.AssociationConf{Version: versions[i]})
+		}
+		return apm
+	}
+	type args struct {
+		resourceVersion version.Version
+		associated      commonv1.Associated
+	}
+	tests := []struct {
+		name string
+		args args
+		want bool
+	}{
+		{
+			name: "no association specified: allow",
+			args: args{
+				resourceVersion: version.MustParse("7.7.0"),
+				associated:      apmNoAssoc.DeepCopy(),
+			},
+			want: true,
+		},
+		{
+			name: "referenced resources run the same version: allow",
+			args: args{
+				resourceVersion: version.MustParse("7.7.0"),
+				associated:      apmTwoAssocWithVersions([]string{"7.7.0", "7.7.0"}),
+			},
+			want: true,
+		},
+		{
+			name: "some referenced resources run a higher version: allow",
+			args: args{
+				resourceVersion: version.MustParse("7.7.0"),
+				associated:      apmTwoAssocWithVersions([]string{"7.8.0", "7.7.0"}),
+			},
+			want: true,
+		},
+		{
+			name: "one referenced resource runs a lower version: don't allow",
+			args: args{
+				resourceVersion: version.MustParse("7.7.0"),
+				associated:      apmTwoAssocWithVersions([]string{"7.7.0", "7.6.0"}),
+			},
+			want: false,
+		},
+		{
+			name: "no version set in the association conf: don't allow",
+			args: args{
+				resourceVersion: version.MustParse("7.7.0"),
+				associated:      apmTwoAssocWithVersions([]string{"", ""}),
+			},
+			want: false,
+		},
+		{
+			name: "association conf annotation is not set yet: don't allow",
+			args: args{
+				resourceVersion: version.MustParse("7.7.0"),
+				associated:      apmTwoAssoc,
+			},
+			want: false,
+		},
+		{
+			name: "invalid version in the association conf: don't allow",
+			args: args{
+				resourceVersion: version.MustParse("7.7.0"),
+				associated:      apmTwoAssocWithVersions([]string{"7.7.0", "invalid"}),
+			},
+			want: false,
+		},
+	}
+	for _, tt := range tests {
+		logger := log.WithValues("a", "b")
+		t.Run(tt.name, func(t *testing.T) {
+			if got := AllowVersion(tt.args.resourceVersion, tt.args.associated, logger); got != tt.want {
+				t.Errorf("AllowVersion() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

pkg/controller/association/controller/apm_es.go

@@ -39,9 +39,10 @@ func AddApmES(mgr manager.Manager, accessReviewer rbac.AccessReviewer, params op
 		ElasticsearchRef: func(c k8s.Client, association commonv1.Association) (bool, commonv1.ObjectSelector, error) {
 			return true, association.AssociationRef(), nil
 		},
-		ExternalServiceURL: getElasticsearchExternalURL,
-		AssociatedNamer:    esv1.ESNamer,
-		AssociationName:    "apm-es",
+		ReferencedResourceVersion: referencedElasticsearchStatusVersion,
+		ExternalServiceURL:        getElasticsearchExternalURL,
+		AssociatedNamer:           esv1.ESNamer,
+		AssociationName:           "apm-es",
 		AssociationLabels: func(associated types.NamespacedName) map[string]string {
 			return map[string]string{
 				ApmAssociationLabelName:      associated.Name,
@@ -67,6 +68,16 @@ func getElasticsearchExternalURL(c k8s.Client, association commonv1.Association)
 	return services.ExternalServiceURL(es), nil
 }
 
+// referencedElasticsearchStatusVersion returns the currently running version of Elasticsearch
+// reported in its status.
+func referencedElasticsearchStatusVersion(c k8s.Client, esRef types.NamespacedName) (string, error) {
+	var es esv1.Elasticsearch
+	if err := c.Get(esRef, &es); err != nil {
+		return "", err
+	}
+	return es.Status.Version, nil
+}
+
 // getAPMElasticsearchRoles returns for a given version of the APM Server the set of required roles.
 func getAPMElasticsearchRoles(associated commonv1.Associated) (string, error) {
 	apmServer, ok := associated.(*apmv1.ApmServer)

pkg/controller/association/controller/apm_kibana.go

@@ -25,12 +25,13 @@ import (
 
 func AddApmKibana(mgr manager.Manager, accessReviewer rbac.AccessReviewer, params operator.Parameters) error {
 	return association.AddAssociationController(mgr, accessReviewer, params, association.AssociationInfo{
-		AssociatedShortName:    "apm",
-		AssociationObjTemplate: func() commonv1.Association { return &apmv1.ApmKibanaAssociation{} },
-		ExternalServiceURL:     getKibanaExternalURL,
-		ElasticsearchRef:       getElasticsearchFromKibana,
-		AssociatedNamer:        kibana.Namer,
-		AssociationName:        "apm-kibana",
+		AssociatedShortName:       "apm",
+		AssociationObjTemplate:    func() commonv1.Association { return &apmv1.ApmKibanaAssociation{} },
+		ExternalServiceURL:        getKibanaExternalURL,
+		ReferencedResourceVersion: referencedKibanaStatusVersion,
+		ElasticsearchRef:          getElasticsearchFromKibana,
+		AssociatedNamer:           kibana.Namer,
+		AssociationName:           "apm-kibana",
 		AssociationLabels: func(associated types.NamespacedName) map[string]string {
 			return map[string]string{
 				ApmAssociationLabelName:      associated.Name,
@@ -74,6 +75,16 @@ func getKibanaExternalURL(c k8s.Client, association commonv1.Association) (strin
 	return stringsutil.Concat(kb.Spec.HTTP.Protocol(), "://", kibana.HTTPService(kb.Name), ".", kb.Namespace, ".svc:", strconv.Itoa(kibana.HTTPPort)), nil
 }
 
+// referencedKibanaStatusVersion returns the currently running version of Kibana
+// reported in its status.
+func referencedKibanaStatusVersion(c k8s.Client, kbRef types.NamespacedName) (string, error) {
+	var kb kbv1.Kibana
+	if err := c.Get(kbRef, &kb); err != nil {
+		return "", err
+	}
+	return kb.Status.Version, nil
+}
+
 // getElasticsearchFromKibana returns the Elasticsearch reference in which the user must be created for this association.
 func getElasticsearchFromKibana(c k8s.Client, association commonv1.Association) (bool, commonv1.ObjectSelector, error) {
 	kibanaRef := association.AssociationRef()

pkg/controller/association/controller/beat_es.go

@@ -39,10 +39,11 @@ func AddBeatES(mgr manager.Manager, accessReviewer rbac.AccessReviewer, params o
 		ElasticsearchRef: func(c k8s.Client, association commonv1.Association) (bool, commonv1.ObjectSelector, error) {
 			return true, association.AssociationRef(), nil
 		},
-		ExternalServiceURL:  getElasticsearchExternalURL,
-		AssociatedNamer:     esv1.ESNamer,
-		AssociationName:     "beat-es",
-		AssociatedShortName: "beat",
+		ReferencedResourceVersion: referencedElasticsearchStatusVersion,
+		ExternalServiceURL:        getElasticsearchExternalURL,
+		AssociatedNamer:           esv1.ESNamer,
+		AssociationName:           "beat-es",
+		AssociatedShortName:       "beat",
 		AssociationLabels: func(associated types.NamespacedName) map[string]string {
 			return map[string]string{
 				BeatAssociationLabelName:      associated.Name,

pkg/controller/association/controller/beat_kibana.go

@@ -25,12 +25,13 @@ import (
 
 func AddBeatKibana(mgr manager.Manager, accessReviewer rbac.AccessReviewer, params operator.Parameters) error {
 	return association.AddAssociationController(mgr, accessReviewer, params, association.AssociationInfo{
-		AssociationObjTemplate: func() commonv1.Association { return &beatv1beta1.BeatKibanaAssociation{} },
-		ElasticsearchRef:       getElasticsearchFromKibana,
-		ExternalServiceURL:     getKibanaExternalURL,
-		AssociatedNamer:        kibana.Namer,
-		AssociationName:        "beat-kibana",
-		AssociatedShortName:    "beat",
+		AssociationObjTemplate:    func() commonv1.Association { return &beatv1beta1.BeatKibanaAssociation{} },
+		ElasticsearchRef:          getElasticsearchFromKibana,
+		ExternalServiceURL:        getKibanaExternalURL,
+		ReferencedResourceVersion: referencedKibanaStatusVersion,
+		AssociatedNamer:           kibana.Namer,
+		AssociationName:           "beat-kibana",
+		AssociatedShortName:       "beat",
 		AssociationLabels: func(associated types.NamespacedName) map[string]string {
 			return map[string]string{
 				BeatAssociationLabelName:      associated.Name,

pkg/controller/association/controller/ent_es.go

@@ -5,6 +5,9 @@
 package controller
 
 import (
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+
 	commonv1 "github.com/elastic/cloud-on-k8s/pkg/apis/common/v1"
 	esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
 	entv1beta1 "github.com/elastic/cloud-on-k8s/pkg/apis/enterprisesearch/v1beta1"
@@ -14,8 +17,6 @@ import (
 	esuser "github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/user"
 	"github.com/elastic/cloud-on-k8s/pkg/utils/k8s"
 	"github.com/elastic/cloud-on-k8s/pkg/utils/rbac"
-	"k8s.io/apimachinery/pkg/types"
-	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
 const (
@@ -31,10 +32,11 @@ func AddEntES(mgr manager.Manager, accessReviewer rbac.AccessReviewer, params op
 		ElasticsearchRef: func(c k8s.Client, association commonv1.Association) (bool, commonv1.ObjectSelector, error) {
 			return true, association.AssociationRef(), nil
 		},
-		ExternalServiceURL:  getElasticsearchExternalURL,
-		AssociatedNamer:     esv1.ESNamer,
-		AssociationName:     "ent-es",
-		AssociatedShortName: "ent",
+		ReferencedResourceVersion: referencedElasticsearchStatusVersion,
+		ExternalServiceURL:        getElasticsearchExternalURL,
+		AssociatedNamer:           esv1.ESNamer,
+		AssociationName:           "ent-es",
+		AssociatedShortName:       "ent",
 		AssociationLabels: func(associated types.NamespacedName) map[string]string {
 			return map[string]string{
 				EntESAssociationLabelName:      associated.Name,

pkg/controller/association/controller/kibana_es.go

@@ -5,6 +5,9 @@
 package controller
 
 import (
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+
 	commonv1 "github.com/elastic/cloud-on-k8s/pkg/apis/common/v1"
 	esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
 	kbv1 "github.com/elastic/cloud-on-k8s/pkg/apis/kibana/v1"
@@ -13,8 +16,6 @@ import (
 	eslabel "github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/label"
 	"github.com/elastic/cloud-on-k8s/pkg/utils/k8s"
 	"github.com/elastic/cloud-on-k8s/pkg/utils/rbac"
-	"k8s.io/apimachinery/pkg/types"
-	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
 const (
@@ -33,10 +34,11 @@ func AddKibanaES(mgr manager.Manager, accessReviewer rbac.AccessReviewer, params
 		ElasticsearchRef: func(c k8s.Client, association commonv1.Association) (bool, commonv1.ObjectSelector, error) {
 			return true, association.AssociationRef(), nil
 		},
-		ExternalServiceURL:  getElasticsearchExternalURL,
-		AssociatedNamer:     esv1.ESNamer,
-		AssociationName:     "kb-es",
-		AssociatedShortName: "kb",
+		ReferencedResourceVersion: referencedElasticsearchStatusVersion,
+		ExternalServiceURL:        getElasticsearchExternalURL,
+		AssociatedNamer:           esv1.ESNamer,
+		AssociationName:           "kb-es",
+		AssociatedShortName:       "kb",
 		AssociationLabels: func(associated types.NamespacedName) map[string]string {
 			return map[string]string{
 				KibanaESAssociationLabelName:      associated.Name,