diff --git a/hack/cluster-version-util/task_graph.go b/hack/cluster-version-util/task_graph.go
index dd17178ed9..6fe47e2059 100644
--- a/hack/cluster-version-util/task_graph.go
+++ b/hack/cluster-version-util/task_graph.go
@@ -7,6 +7,7 @@ import (
 
 	"github.com/spf13/cobra"
 
+	"github.com/openshift/cluster-version-operator/lib/capability"
 	"github.com/openshift/cluster-version-operator/pkg/payload"
 )
 
@@ -30,7 +31,7 @@ func newTaskGraphCmd() *cobra.Command {
 
 func runTaskGraphCmd(cmd *cobra.Command, args []string) error {
 	manifestDir := args[0]
-	release, err := payload.LoadUpdate(manifestDir, "", "", false, payload.DefaultClusterProfile)
+	release, err := payload.LoadUpdate(manifestDir, "", "", false, payload.DefaultClusterProfile, capability.ClusterCapabilities{})
 	if err != nil {
 		return err
 	}
diff --git a/lib/capability/capability.go b/lib/capability/capability.go
new file mode 100644
index 0000000000..83cb2a55ce
--- /dev/null
+++ b/lib/capability/capability.go
@@ -0,0 +1,111 @@
+package capability
+
+import (
+	"fmt"
+	"strings"
+
+	configv1 "github.com/openshift/api/config/v1"
+)
+
+const (
+	CapabilityAnnotation = "capability.openshift.io/name"
+
+	DefaultCapabilitySet = configv1.ClusterVersionCapabilitySetCurrent
+)
+
+type ClusterCapabilities struct {
+	KnownCapabilities   map[configv1.ClusterVersionCapability]struct{}
+	EnabledCapabilities map[configv1.ClusterVersionCapability]struct{}
+}
+
+// SetCapabilities populates and returns cluster capabilities from ClusterVersion capabilities spec.
+func SetCapabilities(config *configv1.ClusterVersion) ClusterCapabilities {
+	var capabilities ClusterCapabilities
+	capabilities.KnownCapabilities = setKnownCapabilities(config)
+	capabilities.EnabledCapabilities = setEnabledCapabilities(config)
+	return capabilities
+}
+
+// GetCapabilitiesStatus populates and returns ClusterVersion capabilities status from given capabilities.
+func GetCapabilitiesStatus(capabilities ClusterCapabilities) configv1.ClusterVersionCapabilitiesStatus {
+	var status configv1.ClusterVersionCapabilitiesStatus
+	for k := range capabilities.EnabledCapabilities {
+		status.EnabledCapabilities = append(status.EnabledCapabilities, k)
+	}
+	for k := range capabilities.KnownCapabilities {
+		status.KnownCapabilities = append(status.KnownCapabilities, k)
+	}
+	return status
+}
+
+// CheckResourceEnablement, given resource annotations and defined cluster capabilities, checks if the capability
+// annotation exists. If so, each capability name is validated against the known set of capabilities. Each valid
+// capability is then checked if it is disabled. If any invalid capabilities are found an error is returned listing
+// all invalid capabilities. Otherwise, if any disabled capabilities are found an error is returned listing all
+// disabled capabilities.
+func CheckResourceEnablement(annotations map[string]string, capabilities ClusterCapabilities) error {
+	val, ok := annotations[CapabilityAnnotation]
+	if !ok {
+		return nil
+	}
+	caps := strings.Split(val, "+")
+	numCaps := len(caps)
+	unknownCaps := make([]string, 0, numCaps)
+	disabledCaps := make([]string, 0, numCaps)
+
+	for _, c := range caps {
+		if _, ok = capabilities.KnownCapabilities[configv1.ClusterVersionCapability(c)]; !ok {
+			unknownCaps = append(unknownCaps, c)
+		} else if _, ok = capabilities.EnabledCapabilities[configv1.ClusterVersionCapability(c)]; !ok {
+			disabledCaps = append(disabledCaps, c)
+		}
+	}
+	if len(unknownCaps) > 0 {
+		return fmt.Errorf("unrecognized capability names: %s", strings.Join(unknownCaps, ", "))
+	}
+	if len(disabledCaps) > 0 {
+		return fmt.Errorf("disabled capabilities: %s", strings.Join(disabledCaps, ", "))
+	}
+	return nil
+}
+
+// setKnownCapabilities populates a map keyed by capability from all known capabilities as defined in ClusterVersion.
+func setKnownCapabilities(config *configv1.ClusterVersion) map[configv1.ClusterVersionCapability]struct{} {
+	known := make(map[configv1.ClusterVersionCapability]struct{})
+
+	for _, v := range configv1.ClusterVersionCapabilitySets {
+		for _, capability := range v {
+			if _, ok := known[capability]; ok {
+				continue
+			}
+			known[capability] = struct{}{}
+		}
+	}
+	return known
+}
+
+// setEnabledCapabilities populates a map keyed by capability from all enabled capabilities as defined in ClusterVersion.
+// DefaultCapabilitySet is used if a baseline capability set is not defined by ClusterVersion.
+func setEnabledCapabilities(config *configv1.ClusterVersion) map[configv1.ClusterVersionCapability]struct{} {
+	enabled := make(map[configv1.ClusterVersionCapability]struct{})
+
+	capSet := DefaultCapabilitySet
+
+	if config.Spec.Capabilities != nil && len(config.Spec.Capabilities.BaselineCapabilitySet) > 0 {
+		capSet = config.Spec.Capabilities.BaselineCapabilitySet
+	}
+
+	for _, v := range configv1.ClusterVersionCapabilitySets[capSet] {
+		enabled[v] = struct{}{}
+	}
+
+	if config.Spec.Capabilities != nil {
+		for _, v := range config.Spec.Capabilities.AdditionalEnabledCapabilities {
+			if _, ok := enabled[v]; ok {
+				continue
+			}
+			enabled[v] = struct{}{}
+		}
+	}
+	return enabled
+}
diff --git a/lib/capability/capability_test.go b/lib/capability/capability_test.go
new file mode 100644
index 0000000000..94955323f3
--- /dev/null
+++ b/lib/capability/capability_test.go
@@ -0,0 +1,287 @@
+package capability
+
+import (
+	"testing"
+
+	configv1 "github.com/openshift/api/config/v1"
+)
+
+func TestSetCapabilities(t *testing.T) {
+	tests := []struct {
+		name            string
+		config          *configv1.ClusterVersion
+		wantKnownKeys   []string
+		wantEnabledKeys []string
+	}{
+		{name: "capabilities nil",
+			config: &configv1.ClusterVersion{},
+			// wantKnownKeys and wantEnabledKeys will be set to default set of capabilities by test
+		},
+		{name: "capabilities set not set",
+			config: &configv1.ClusterVersion{
+				Spec: configv1.ClusterVersionSpec{
+					Capabilities: &configv1.ClusterVersionCapabilitiesSpec{},
+				},
+			},
+			// wantKnownKeys and wantEnabledKeys will be set to default set of capabilities by test
+		},
+		{name: "set capabilities None",
+			config: &configv1.ClusterVersion{
+				Spec: configv1.ClusterVersionSpec{
+					Capabilities: &configv1.ClusterVersionCapabilitiesSpec{
+						BaselineCapabilitySet: configv1.ClusterVersionCapabilitySetNone,
+					},
+				},
+			},
+			wantKnownKeys:   []string{string(configv1.ClusterVersionCapabilityOpenShiftSamples)},
+			wantEnabledKeys: []string{},
+		},
+		{name: "set capabilities 4_11",
+			config: &configv1.ClusterVersion{
+				Spec: configv1.ClusterVersionSpec{
+					Capabilities: &configv1.ClusterVersionCapabilitiesSpec{
+						BaselineCapabilitySet:         configv1.ClusterVersionCapabilitySet4_11,
+						AdditionalEnabledCapabilities: []configv1.ClusterVersionCapability{},
+					},
+				},
+			},
+			wantKnownKeys:   []string{string(configv1.ClusterVersionCapabilityOpenShiftSamples)},
+			wantEnabledKeys: []string{string(configv1.ClusterVersionCapabilityOpenShiftSamples)},
+		},
+		{name: "set capabilities vCurrent",
+			config: &configv1.ClusterVersion{
+				Spec: configv1.ClusterVersionSpec{
+					Capabilities: &configv1.ClusterVersionCapabilitiesSpec{
+						BaselineCapabilitySet:         configv1.ClusterVersionCapabilitySetCurrent,
+						AdditionalEnabledCapabilities: []configv1.ClusterVersionCapability{},
+					},
+				},
+			},
+			wantKnownKeys:   []string{string(configv1.ClusterVersionCapabilityOpenShiftSamples)},
+			wantEnabledKeys: []string{string(configv1.ClusterVersionCapabilityOpenShiftSamples)},
+		},
+		{name: "set capabilities None with additional",
+			config: &configv1.ClusterVersion{
+				Spec: configv1.ClusterVersionSpec{
+					Capabilities: &configv1.ClusterVersionCapabilitiesSpec{
+						BaselineCapabilitySet:         configv1.ClusterVersionCapabilitySetNone,
+						AdditionalEnabledCapabilities: []configv1.ClusterVersionCapability{"cap1", "cap2", "cap3"},
+					},
+				},
+			},
+			wantKnownKeys:   []string{string(configv1.ClusterVersionCapabilityOpenShiftSamples)},
+			wantEnabledKeys: []string{"cap1", "cap2", "cap3"},
+		},
+		{name: "set capabilities 4_11 with additional",
+			config: &configv1.ClusterVersion{
+				Spec: configv1.ClusterVersionSpec{
+					Capabilities: &configv1.ClusterVersionCapabilitiesSpec{
+						BaselineCapabilitySet:         configv1.ClusterVersionCapabilitySet4_11,
+						AdditionalEnabledCapabilities: []configv1.ClusterVersionCapability{"cap1", "cap2", "cap3"},
+					},
+				},
+			},
+			wantKnownKeys:   []string{string(configv1.ClusterVersionCapabilityOpenShiftSamples)},
+			wantEnabledKeys: []string{string(configv1.ClusterVersionCapabilityOpenShiftSamples), "cap1", "cap2", "cap3"},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			caps := SetCapabilities(test.config)
+			if test.config.Spec.Capabilities == nil || (test.config.Spec.Capabilities != nil &&
+				len(test.config.Spec.Capabilities.BaselineCapabilitySet) == 0) {
+
+				test.wantKnownKeys = getDefaultCapabilities()
+				test.wantEnabledKeys = getDefaultCapabilities()
+			}
+			if len(caps.KnownCapabilities) != len(test.wantKnownKeys) {
+				t.Errorf("Incorrect number of KnownCapabilities keys, wanted: %q. KnownCapabilities returned: %v", test.wantKnownKeys, caps.KnownCapabilities)
+			}
+			for _, v := range test.wantKnownKeys {
+				if _, ok := caps.KnownCapabilities[configv1.ClusterVersionCapability(v)]; !ok {
+					t.Errorf("Missing KnownCapabilities key %q. KnownCapabilities returned : %v", v, caps.KnownCapabilities)
+				}
+			}
+			if len(caps.EnabledCapabilities) != len(test.wantEnabledKeys) {
+				t.Errorf("Incorrect number of EnabledCapabilities keys, wanted: %q. EnabledCapabilities returned: %v", test.wantEnabledKeys, caps.EnabledCapabilities)
+			}
+			for _, v := range test.wantEnabledKeys {
+				if _, ok := caps.EnabledCapabilities[configv1.ClusterVersionCapability(v)]; !ok {
+					t.Errorf("Missing EnabledCapabilities key %q. EnabledCapabilities returned : %v", v, caps.EnabledCapabilities)
+				}
+			}
+		})
+	}
+}
+
+func TestGetCapabilitiesStatus(t *testing.T) {
+	tests := []struct {
+		name       string
+		caps       ClusterCapabilities
+		wantStatus configv1.ClusterVersionCapabilitiesStatus
+	}{
+		{name: "empty capabilities",
+			caps: ClusterCapabilities{
+				KnownCapabilities:   map[configv1.ClusterVersionCapability]struct{}{},
+				EnabledCapabilities: map[configv1.ClusterVersionCapability]struct{}{},
+			},
+			wantStatus: configv1.ClusterVersionCapabilitiesStatus{
+				EnabledCapabilities: []configv1.ClusterVersionCapability{},
+				KnownCapabilities:   []configv1.ClusterVersionCapability{},
+			},
+		},
+		{name: "capabilities",
+			caps: ClusterCapabilities{
+				KnownCapabilities:   map[configv1.ClusterVersionCapability]struct{}{configv1.ClusterVersionCapabilityOpenShiftSamples: {}},
+				EnabledCapabilities: map[configv1.ClusterVersionCapability]struct{}{configv1.ClusterVersionCapabilityOpenShiftSamples: {}},
+			},
+			wantStatus: configv1.ClusterVersionCapabilitiesStatus{
+				EnabledCapabilities: []configv1.ClusterVersionCapability{configv1.ClusterVersionCapabilityOpenShiftSamples},
+				KnownCapabilities:   []configv1.ClusterVersionCapability{configv1.ClusterVersionCapabilityOpenShiftSamples},
+			},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			config := GetCapabilitiesStatus(test.caps)
+			if len(config.KnownCapabilities) != len(test.wantStatus.KnownCapabilities) {
+				t.Errorf("Incorrect number of KnownCapabilities keys, wanted: %q. KnownCapabilities returned: %v",
+					test.wantStatus.KnownCapabilities, config.KnownCapabilities)
+			}
+			for _, v := range test.wantStatus.KnownCapabilities {
+				vFound := false
+				for _, cv := range config.KnownCapabilities {
+					if v == cv {
+						vFound = true
+						break
+					}
+					if !vFound {
+						t.Errorf("Missing KnownCapabilities key %q. KnownCapabilities returned : %v", v, config.KnownCapabilities)
+					}
+				}
+			}
+			if len(config.EnabledCapabilities) != len(test.wantStatus.EnabledCapabilities) {
+				t.Errorf("Incorrect number of EnabledCapabilities keys, wanted: %q. EnabledCapabilities returned: %v",
+					test.wantStatus.EnabledCapabilities, config.EnabledCapabilities)
+			}
+			for _, v := range test.wantStatus.EnabledCapabilities {
+				vFound := false
+				for _, cv := range config.EnabledCapabilities {
+					if v == cv {
+						vFound = true
+						break
+					}
+					if !vFound {
+						t.Errorf("Missing EnabledCapabilities key %q. EnabledCapabilities returned : %v", v, config.EnabledCapabilities)
+					}
+				}
+			}
+		})
+	}
+}
+
+func TestCheckResourceEnablement(t *testing.T) {
+	tests := []struct {
+		name        string
+		annotations map[string]string
+		caps        ClusterCapabilities
+		wantError   string
+	}{
+		{name: "empty annotations"},
+		{name: "no capabilitity annotation",
+			annotations: map[string]string{"foo": "bar"},
+		},
+		{name: "known capabilitity annotation",
+			annotations: map[string]string{CapabilityAnnotation: string(configv1.ClusterVersionCapabilityOpenShiftSamples), "foo": "bar"},
+			caps: ClusterCapabilities{
+				KnownCapabilities:   map[configv1.ClusterVersionCapability]struct{}{configv1.ClusterVersionCapabilityOpenShiftSamples: {}},
+				EnabledCapabilities: map[configv1.ClusterVersionCapability]struct{}{configv1.ClusterVersionCapabilityOpenShiftSamples: {}},
+			},
+		},
+		{name: "multiple enabled capabilitities annotation",
+			annotations: map[string]string{CapabilityAnnotation: "cap1+cap2"},
+			caps: ClusterCapabilities{
+				KnownCapabilities:   map[configv1.ClusterVersionCapability]struct{}{"cap1": {}, "cap2": {}},
+				EnabledCapabilities: map[configv1.ClusterVersionCapability]struct{}{"cap1": {}, "cap2": {}},
+			},
+		},
+		{name: "multiple capabilitities annotation with spaces",
+			annotations: map[string]string{CapabilityAnnotation: " + cap1 +cap2+ "},
+			caps: ClusterCapabilities{
+				KnownCapabilities:   map[configv1.ClusterVersionCapability]struct{}{"cap1": {}, "cap2": {}},
+				EnabledCapabilities: map[configv1.ClusterVersionCapability]struct{}{"cap1": {}, "cap2": {}},
+			},
+			wantError: "unrecognized capability names:  ,  cap1 ,  ",
+		},
+		{name: "unrecognized capabilitity annotation",
+			annotations: map[string]string{CapabilityAnnotation: "cap1"},
+			caps: ClusterCapabilities{
+				KnownCapabilities:   map[configv1.ClusterVersionCapability]struct{}{configv1.ClusterVersionCapabilityOpenShiftSamples: {}},
+				EnabledCapabilities: map[configv1.ClusterVersionCapability]struct{}{configv1.ClusterVersionCapabilityOpenShiftSamples: {}},
+			},
+			wantError: "unrecognized capability names: cap1",
+		},
+		{name: "unrecognized capabilitities, spaces",
+			annotations: map[string]string{CapabilityAnnotation: "cap1 + cap2"},
+			caps: ClusterCapabilities{
+				KnownCapabilities:   map[configv1.ClusterVersionCapability]struct{}{"cap1": {}, "cap2": {}},
+				EnabledCapabilities: map[configv1.ClusterVersionCapability]struct{}{"cap1": {}, "cap2": {}},
+			},
+			wantError: "unrecognized capability names: cap1 ,  cap2",
+		},
+		{name: "invalid capabilitity annotation divider",
+			annotations: map[string]string{CapabilityAnnotation: "cap1,cap2,cap3,cap4"},
+			caps: ClusterCapabilities{
+				KnownCapabilities:   map[configv1.ClusterVersionCapability]struct{}{configv1.ClusterVersionCapabilityOpenShiftSamples: {}},
+				EnabledCapabilities: map[configv1.ClusterVersionCapability]struct{}{configv1.ClusterVersionCapabilityOpenShiftSamples: {}},
+			},
+			wantError: "unrecognized capability names: cap1,cap2,cap3,cap4",
+		},
+		{name: "multiple unrecognized capabilitities annotation",
+			annotations: map[string]string{CapabilityAnnotation: "cap1+cap2+cap3+cap4"},
+			caps: ClusterCapabilities{
+				KnownCapabilities:   map[configv1.ClusterVersionCapability]struct{}{configv1.ClusterVersionCapabilityOpenShiftSamples: {}},
+				EnabledCapabilities: map[configv1.ClusterVersionCapability]struct{}{configv1.ClusterVersionCapabilityOpenShiftSamples: {}},
+			},
+			wantError: "unrecognized capability names: cap1, cap2, cap3, cap4",
+		},
+		{name: "disabled capabilitity annotation",
+			annotations: map[string]string{CapabilityAnnotation: "cap1"},
+			caps: ClusterCapabilities{
+				KnownCapabilities:   map[configv1.ClusterVersionCapability]struct{}{"cap1": {}},
+				EnabledCapabilities: map[configv1.ClusterVersionCapability]struct{}{},
+			},
+			wantError: "disabled capabilities: cap1",
+		},
+		{name: "multiple disabled capabilitities annotation",
+			annotations: map[string]string{CapabilityAnnotation: "cap1+cap2+cap3+cap4+cap5+cap6"},
+			caps: ClusterCapabilities{
+				KnownCapabilities: map[configv1.ClusterVersionCapability]struct{}{"cap1": {}, "cap2": {},
+					"cap3": {}, "cap4": {}, "cap5": {}, "cap6": {}},
+				EnabledCapabilities: map[configv1.ClusterVersionCapability]struct{}{"cap1": {}, "cap2": {}, "cap3": {}},
+			},
+			wantError: "disabled capabilities: cap4, cap5, cap6",
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			err := CheckResourceEnablement(test.annotations, test.caps)
+			if len(test.wantError) == 0 {
+				if err != nil {
+					t.Errorf("Wanted no error, got error %q", err.Error())
+				}
+			} else if test.wantError != err.Error() {
+				t.Errorf("Wanted error %q, got error %q", test.wantError, err.Error())
+			}
+		})
+	}
+}
+
+func getDefaultCapabilities() []string {
+	var caps []string
+
+	for _, v := range configv1.ClusterVersionCapabilitySets[DefaultCapabilitySet] {
+		caps = append(caps, string(v))
+	}
+	return caps
+}
diff --git a/pkg/cvo/cvo.go b/pkg/cvo/cvo.go
index c110ea97b3..1a48c79de5 100644
--- a/pkg/cvo/cvo.go
+++ b/pkg/cvo/cvo.go
@@ -30,6 +30,7 @@ import (
 	clientset "github.com/openshift/client-go/config/clientset/versioned"
 	configinformersv1 "github.com/openshift/client-go/config/informers/externalversions/config/v1"
 	configlistersv1 "github.com/openshift/client-go/config/listers/config/v1"
+	"github.com/openshift/cluster-version-operator/lib/capability"
 	"github.com/openshift/cluster-version-operator/lib/resourcebuilder"
 	"github.com/openshift/cluster-version-operator/lib/validation"
 	cvointernal "github.com/openshift/cluster-version-operator/pkg/cvo/internal"
@@ -221,11 +222,14 @@ func New(
 // payload appears to be in error rather than continuing.
 func (optr *Operator) InitializeFromPayload(ctx context.Context, restConfig *rest.Config, burstRestConfig *rest.Config) error {
 
+	var config *configv1.ClusterVersion
+
 	// wait until cluster version object exists
 	if err := wait.PollImmediateInfiniteWithContext(ctx, 3*time.Second, func(ctx context.Context) (bool, error) {
+		var err error
 
 		// ensure the cluster version exists
-		_, _, err := optr.getClusterVersion(ctx)
+		config, _, err = optr.getClusterVersion(ctx)
 		if err != nil {
 			if apierrors.IsNotFound(err) {
 				klog.V(2).Infof("No cluster version object, waiting for one")
@@ -237,8 +241,11 @@ func (optr *Operator) InitializeFromPayload(ctx context.Context, restConfig *res
 	}); err != nil {
 		return fmt.Errorf("Error when attempting to get cluster version object: %w", err)
 	}
+	capabilities := capability.SetCapabilities(config)
+
+	update, err := payload.LoadUpdate(optr.defaultPayloadDir(), optr.release.Image, optr.exclude, optr.includeTechPreview,
+		optr.clusterProfile, capabilities)
 
-	update, err := payload.LoadUpdate(optr.defaultPayloadDir(), optr.release.Image, optr.exclude, optr.includeTechPreview, optr.clusterProfile)
 	if err != nil {
 		return fmt.Errorf("the local release contents are invalid - no current version can be determined from disk: %v", err)
 	}
@@ -578,7 +585,7 @@ func (optr *Operator) sync(ctx context.Context, key string) error {
 	}
 
 	// inform the config sync loop about our desired state
-	status := optr.configSync.Update(ctx, config.Generation, desired, config.Spec.Overrides, state, optr.name)
+	status := optr.configSync.Update(ctx, config.Generation, desired, config, state, optr.name)
 
 	// write cluster version status
 	return optr.syncStatus(ctx, original, config, status, errs)
diff --git a/pkg/cvo/cvo_scenarios_test.go b/pkg/cvo/cvo_scenarios_test.go
index f0a0ce7a55..aa22d2e13d 100644
--- a/pkg/cvo/cvo_scenarios_test.go
+++ b/pkg/cvo/cvo_scenarios_test.go
@@ -207,6 +207,10 @@ func TestCVO_StartupAndSync(t *testing.T) {
 			History: []configv1.UpdateHistory{
 				{State: configv1.PartialUpdate, Image: "image/image:1", Version: "1.0.0-abc", StartedTime: defaultStartedTime},
 			},
+			Capabilities: configv1.ClusterVersionCapabilitiesStatus{
+				EnabledCapabilities: []configv1.ClusterVersionCapability{"openshift-samples"},
+				KnownCapabilities:   []configv1.ClusterVersionCapability{"openshift-samples"},
+			},
 			Conditions: []configv1.ClusterOperatorStatusCondition{
 				{Type: configv1.OperatorAvailable, Status: configv1.ConditionFalse},
 				// cleared failing status and set progressing
@@ -565,6 +569,10 @@ func TestCVO_StartupAndSyncUnverifiedPayload(t *testing.T) {
 			History: []configv1.UpdateHistory{
 				{State: configv1.PartialUpdate, Image: "image/image:1", Version: "1.0.0-abc", StartedTime: defaultStartedTime},
 			},
+			Capabilities: configv1.ClusterVersionCapabilitiesStatus{
+				EnabledCapabilities: []configv1.ClusterVersionCapability{"openshift-samples"},
+				KnownCapabilities:   []configv1.ClusterVersionCapability{"openshift-samples"},
+			},
 			Conditions: []configv1.ClusterOperatorStatusCondition{
 				{Type: configv1.OperatorAvailable, Status: configv1.ConditionFalse},
 				// cleared failing status and set progressing
@@ -900,6 +908,10 @@ func TestCVO_StartupAndSyncPreconditionFailing(t *testing.T) {
 			History: []configv1.UpdateHistory{
 				{State: configv1.PartialUpdate, Image: "image/image:1", Version: "1.0.0-abc", StartedTime: defaultStartedTime},
 			},
+			Capabilities: configv1.ClusterVersionCapabilitiesStatus{
+				EnabledCapabilities: []configv1.ClusterVersionCapability{"openshift-samples"},
+				KnownCapabilities:   []configv1.ClusterVersionCapability{"openshift-samples"},
+			},
 			Conditions: []configv1.ClusterOperatorStatusCondition{
 				{Type: configv1.OperatorAvailable, Status: configv1.ConditionFalse},
 				// cleared failing status and set progressing
@@ -2059,6 +2071,10 @@ func TestCVO_UpgradeVerifiedPayload(t *testing.T) {
 				{State: configv1.PartialUpdate, Image: "image/image:1", Version: "1.0.1-abc", StartedTime: defaultStartedTime},
 				{State: configv1.CompletedUpdate, Image: "image/image:0", Version: "1.0.0-abc", Verified: true, StartedTime: defaultStartedTime, CompletionTime: &defaultCompletionTime},
 			},
+			Capabilities: configv1.ClusterVersionCapabilitiesStatus{
+				EnabledCapabilities: []configv1.ClusterVersionCapability{"openshift-samples"},
+				KnownCapabilities:   []configv1.ClusterVersionCapability{"openshift-samples"},
+			},
 			Conditions: []configv1.ClusterOperatorStatusCondition{
 				{Type: configv1.OperatorAvailable, Status: configv1.ConditionTrue, Message: "Done applying 1.0.0-abc"},
 				// cleared failing status and set progressing
@@ -2236,7 +2252,7 @@ func TestCVO_RestartAndReconcile(t *testing.T) {
 		t.Fatal(err)
 	}
 	actions = client.Actions()
-	if len(actions) != 1 {
+	if len(actions) != 2 {
 		t.Fatalf("%s", spew.Sdump(actions))
 	}
 	expectGet(t, actions[0], "clusterversions", "", "version")
diff --git a/pkg/cvo/status.go b/pkg/cvo/status.go
index 0fdb83fdae..ee711239ab 100644
--- a/pkg/cvo/status.go
+++ b/pkg/cvo/status.go
@@ -201,6 +201,8 @@ func (optr *Operator) syncStatus(ctx context.Context, original, config *configv1
 	desired := optr.mergeReleaseMetadata(status.Actual)
 	mergeOperatorHistory(config, desired, status.Verified, now, status.Completed > 0)
 
+	config.Status.Capabilities = status.CapabilitiesStatus
+
 	// update validation errors
 	var reason string
 	if len(validationErrs) > 0 {
diff --git a/pkg/cvo/sync_test.go b/pkg/cvo/sync_test.go
index 0bf9cbdafd..3b1b92fc2f 100644
--- a/pkg/cvo/sync_test.go
+++ b/pkg/cvo/sync_test.go
@@ -413,7 +413,7 @@ func (r *fakeSyncRecorder) StatusCh() <-chan SyncWorkerStatus {
 func (r *fakeSyncRecorder) Start(ctx context.Context, maxWorkers int, cvoOptrName string, lister configlistersv1.ClusterVersionLister) {
 }
 
-func (r *fakeSyncRecorder) Update(ctx context.Context, generation int64, desired configv1.Update, overrides []configv1.ComponentOverride, state payload.State, cvoOptrName string) *SyncWorkerStatus {
+func (r *fakeSyncRecorder) Update(ctx context.Context, generation int64, desired configv1.Update, config *configv1.ClusterVersion, state payload.State, cvoOptrName string) *SyncWorkerStatus {
 	r.Updates = append(r.Updates, desired)
 	return r.Returns
 }
diff --git a/pkg/cvo/sync_worker.go b/pkg/cvo/sync_worker.go
index 7f55901b16..7fbb9c840f 100644
--- a/pkg/cvo/sync_worker.go
+++ b/pkg/cvo/sync_worker.go
@@ -22,6 +22,7 @@ import (
 
 	configv1 "github.com/openshift/api/config/v1"
 
+	"github.com/openshift/cluster-version-operator/lib/capability"
 	"github.com/openshift/cluster-version-operator/pkg/payload"
 	"github.com/openshift/cluster-version-operator/pkg/payload/precondition"
 	"github.com/openshift/library-go/pkg/manifest"
@@ -30,7 +31,7 @@ import (
 // ConfigSyncWorker abstracts how the image is synchronized to the server. Introduced for testing.
 type ConfigSyncWorker interface {
 	Start(ctx context.Context, maxWorkers int, cvoOptrName string, lister configlistersv1.ClusterVersionLister)
-	Update(ctx context.Context, generation int64, desired configv1.Update, overrides []configv1.ComponentOverride, state payload.State, cvoOptrName string) *SyncWorkerStatus
+	Update(ctx context.Context, generation int64, desired configv1.Update, config *configv1.ClusterVersion, state payload.State, cvoOptrName string) *SyncWorkerStatus
 	StatusCh() <-chan SyncWorkerStatus
 }
 
@@ -75,6 +76,8 @@ type SyncWork struct {
 	// Attempt is incremented each time we attempt to sync a payload and reset
 	// when we change Generation/Desired or successfully synchronize.
 	Attempt int
+
+	Capabilities capability.ClusterCapabilities
 }
 
 // Empty returns true if the image is empty for this work.
@@ -113,6 +116,8 @@ type SyncWorkerStatus struct {
 	Verified bool
 
 	loadPayloadStatus LoadPayloadStatus
+
+	CapabilitiesStatus configv1.ClusterVersionCapabilitiesStatus
 }
 
 // DeepCopy copies the worker status.
@@ -264,7 +269,7 @@ func (w *SyncWorker) syncPayload(ctx context.Context, work *SyncWork, reporter S
 		}
 
 		w.eventRecorder.Eventf(cvoObjectRef, corev1.EventTypeNormal, "LoadPayload", "Loading payload version=%q image=%q", desired.Version, desired.Image)
-		payloadUpdate, err := payload.LoadUpdate(info.Directory, desired.Image, w.exclude, w.includeTechPreview, w.clusterProfile)
+		payloadUpdate, err := payload.LoadUpdate(info.Directory, desired.Image, w.exclude, w.includeTechPreview, w.clusterProfile, work.Capabilities)
 		if err != nil {
 			msg := fmt.Sprintf("Loading payload failed version=%q image=%q failure=%v", desired.Version, desired.Image, err)
 			w.eventRecorder.Eventf(cvoObjectRef, corev1.EventTypeWarning, "LoadPayloadFailed", msg)
@@ -362,14 +367,16 @@ func (w *SyncWorker) loadUpdatedPayload(ctx context.Context, work *SyncWork, cvo
 // the initial state or whatever the last recorded status was.
 // TODO: in the future it may be desirable for changes that alter desired to wait briefly before returning,
 //   giving the sync loop the opportunity to observe our change and begin working towards it.
-func (w *SyncWorker) Update(ctx context.Context, generation int64, desired configv1.Update, overrides []configv1.ComponentOverride, state payload.State, cvoOptrName string) *SyncWorkerStatus {
+func (w *SyncWorker) Update(ctx context.Context, generation int64, desired configv1.Update, config *configv1.ClusterVersion,
+	state payload.State, cvoOptrName string) *SyncWorkerStatus {
+
 	w.lock.Lock()
 	defer w.lock.Unlock()
 
 	work := &SyncWork{
 		Generation: generation,
 		Desired:    desired,
-		Overrides:  overrides,
+		Overrides:  config.Spec.Overrides,
 	}
 
 	// the sync worker’s generation should always be latest with every change
@@ -382,6 +389,7 @@ func (w *SyncWorker) Update(ctx context.Context, generation int64, desired confi
 		return w.status.DeepCopy()
 	}
 
+	work.Capabilities = capability.SetCapabilities(config)
 	versionEqual, overridesEqual := equalSyncWork(w.work, work, fmt.Sprintf("considering cluster version generation %d", generation))
 
 	if versionEqual && overridesEqual {
@@ -425,6 +433,8 @@ func (w *SyncWorker) Update(ctx context.Context, generation int64, desired confi
 		return w.status.DeepCopy()
 	}
 
+	w.status.CapabilitiesStatus = capability.GetCapabilitiesStatus(w.work.Capabilities)
+
 	// notify the sync loop that we changed config
 	if w.cancelFn != nil {
 		klog.V(2).Info("Cancel the sync worker's current loop")
diff --git a/pkg/payload/payload.go b/pkg/payload/payload.go
index 32cf5a5f4e..26d45904c2 100644
--- a/pkg/payload/payload.go
+++ b/pkg/payload/payload.go
@@ -23,6 +23,7 @@ import (
 	configv1 "github.com/openshift/api/config/v1"
 	imagev1 "github.com/openshift/api/image/v1"
 
+	"github.com/openshift/cluster-version-operator/lib/capability"
 	"github.com/openshift/cluster-version-operator/lib/resourceread"
 	"github.com/openshift/library-go/pkg/manifest"
 )
@@ -132,7 +133,9 @@ type metadata struct {
 	Metadata map[string]interface{}
 }
 
-func LoadUpdate(dir, releaseImage, excludeIdentifier string, includeTechPreview bool, profile string) (*Update, error) {
+func LoadUpdate(dir, releaseImage, excludeIdentifier string, includeTechPreview bool, profile string,
+	capabilities capability.ClusterCapabilities) (*Update, error) {
+
 	payload, tasks, err := loadUpdatePayloadMetadata(dir, releaseImage, profile)
 	if err != nil {
 		return nil, err
@@ -187,7 +190,7 @@ func LoadUpdate(dir, releaseImage, excludeIdentifier string, includeTechPreview
 			// Filter out manifests that should be excluded based on annotation
 			filteredMs := []manifest.Manifest{}
 			for _, manifest := range ms {
-				if err := include(excludeIdentifier, includeTechPreview, profile, &manifest); err != nil {
+				if err := include(excludeIdentifier, includeTechPreview, profile, capabilities, &manifest); err != nil {
 					klog.V(5).Infof("excluding %s group=%s kind=%s namespace=%s name=%s: %v\n", manifest.OriginalFilename, manifest.GVK.Group, manifest.GVK.Kind, manifest.Obj.GetNamespace(), manifest.Obj.GetName(), err)
 					continue
 				}
@@ -215,7 +218,9 @@ func LoadUpdate(dir, releaseImage, excludeIdentifier string, includeTechPreview
 	return payload, nil
 }
 
-func include(excludeIdentifier string, includeTechPreview bool, profile string, manifest *manifest.Manifest) error {
+func include(excludeIdentifier string, includeTechPreview bool, profile string, capabilities capability.ClusterCapabilities,
+	manifest *manifest.Manifest) error {
+
 	annotations := manifest.Obj.GetAnnotations()
 	if annotations == nil {
 		return errors.New("no annotations")
@@ -237,12 +242,13 @@ func include(excludeIdentifier string, includeTechPreview bool, profile string,
 	}
 
 	profileAnnotation := fmt.Sprintf("include.release.openshift.io/%s", profile)
-	if val, ok := annotations[profileAnnotation]; ok && val == "true" {
-		return nil
-	} else if ok {
+	if val, ok := annotations[profileAnnotation]; ok && val != "true" {
 		return fmt.Errorf("unrecognized value %s=%s", profileAnnotation, val)
+	} else if !ok {
+		return fmt.Errorf("%s unset", profileAnnotation)
 	}
-	return fmt.Errorf("%s unset", profileAnnotation)
+
+	return capability.CheckResourceEnablement(annotations, capabilities)
 }
 
 // ValidateDirectory checks if a directory can be a candidate update by
diff --git a/pkg/payload/payload_test.go b/pkg/payload/payload_test.go
index 61ed9c1fcb..42d55605bb 100644
--- a/pkg/payload/payload_test.go
+++ b/pkg/payload/payload_test.go
@@ -16,6 +16,7 @@ import (
 	configv1 "github.com/openshift/api/config/v1"
 	imagev1 "github.com/openshift/api/image/v1"
 
+	"github.com/openshift/cluster-version-operator/lib/capability"
 	"github.com/openshift/library-go/pkg/manifest"
 )
 
@@ -114,7 +115,7 @@ func TestLoadUpdate(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := LoadUpdate(tt.args.dir, tt.args.releaseImage, "exclude-test", false, DefaultClusterProfile)
+			got, err := LoadUpdate(tt.args.dir, tt.args.releaseImage, "exclude-test", false, DefaultClusterProfile, capability.ClusterCapabilities{})
 			if (err != nil) != tt.wantErr {
 				t.Errorf("loadUpdatePayload() error = %v, wantErr %v", err, tt.wantErr)
 				return
@@ -144,6 +145,7 @@ func Test_include(t *testing.T) {
 		includeTechPreview bool
 		profile            string
 		annotations        map[string]interface{}
+		caps               capability.ClusterCapabilities
 
 		expected error
 	}{
@@ -216,6 +218,36 @@ func Test_include(t *testing.T) {
 			annotations: nil,
 			expected:    errors.New("no annotations"),
 		},
+		{
+			name:    "unrecognized capability works",
+			profile: DefaultClusterProfile,
+			annotations: map[string]interface{}{
+				"include.release.openshift.io/self-managed-high-availability": "true",
+				capability.CapabilityAnnotation:                               "cap1"},
+			expected: errors.New("unrecognized capability names: cap1"),
+		},
+		{
+			name:    "disabled capability works",
+			profile: DefaultClusterProfile,
+			annotations: map[string]interface{}{
+				"include.release.openshift.io/self-managed-high-availability": "true",
+				capability.CapabilityAnnotation:                               "cap1"},
+			caps: capability.ClusterCapabilities{
+				KnownCapabilities: map[configv1.ClusterVersionCapability]struct{}{"cap1": {}},
+			},
+			expected: errors.New("disabled capabilities: cap1"),
+		},
+		{
+			name:    "enabled capability works",
+			profile: DefaultClusterProfile,
+			annotations: map[string]interface{}{
+				"include.release.openshift.io/self-managed-high-availability": "true",
+				capability.CapabilityAnnotation:                               "cap1"},
+			caps: capability.ClusterCapabilities{
+				KnownCapabilities:   map[configv1.ClusterVersionCapability]struct{}{"cap1": {}},
+				EnabledCapabilities: map[configv1.ClusterVersionCapability]struct{}{"cap1": {}},
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -223,7 +255,7 @@ func Test_include(t *testing.T) {
 			if tt.annotations != nil {
 				metadata["annotations"] = tt.annotations
 			}
-			err := include(tt.exclude, tt.includeTechPreview, tt.profile, &manifest.Manifest{
+			err := include(tt.exclude, tt.includeTechPreview, tt.profile, tt.caps, &manifest.Manifest{
 				Obj: &unstructured.Unstructured{
 					Object: map[string]interface{}{
 						"metadata": metadata,
diff --git a/pkg/payload/task_graph_test.go b/pkg/payload/task_graph_test.go
index abe8fb371a..371c9b9547 100644
--- a/pkg/payload/task_graph_test.go
+++ b/pkg/payload/task_graph_test.go
@@ -15,6 +15,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/diff"
 
+	"github.com/openshift/cluster-version-operator/lib/capability"
 	"github.com/openshift/library-go/pkg/manifest"
 )
 
@@ -487,7 +488,7 @@ func Test_TaskGraph_real(t *testing.T) {
 	if len(path) == 0 {
 		t.Skip("TEST_GRAPH_PATH unset")
 	}
-	p, err := LoadUpdate(path, "arbitrary/image:1", "", false, DefaultClusterProfile)
+	p, err := LoadUpdate(path, "arbitrary/image:1", "", false, DefaultClusterProfile, capability.ClusterCapabilities{})
 	if err != nil {
 		t.Fatal(err)
 	}