Skip to content

Commit 05e1615

Browse files
Jeff Peelernjhale
Jeff Peeler
authored and
njhale
committed
feat(catalog): make install possible from bundle image
This extracts bundle data from an image and uses that data to populate the installplan. (The code here depends on later commits.)
1 parent 79bd139 commit 05e1615

File tree

6 files changed

+291
-38
lines changed

6 files changed

+291
-38
lines changed

‎pkg/api/apis/operators/v1alpha1/installplan_types.go

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,11 @@ import (
44
"errors"
55
"fmt"
66

7+
batchv1 "k8s.io/api/batch/v1"
78
corev1 "k8s.io/api/core/v1"
89
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
10+
11+
"github.com/operator-framework/operator-registry/pkg/api"
912
)
1013

1114
const (
@@ -84,6 +87,7 @@ type InstallPlanStatus struct {
8487
Conditions []InstallPlanCondition `json:"conditions,omitempty"`
8588
CatalogSources []string `json:"catalogSources"`
8689
Plan []*Step `json:"plan,omitempty"`
90+
BundleLookups []*BundleLookup `json:"bundleLookup,omitempty"`
8791

8892
// AttenuatedServiceAccountRef references the service account that is used
8993
// to do scoped operator install.
@@ -162,6 +166,24 @@ type Step struct {
162166
Status StepStatus `json:"status"`
163167
}
164168

169+
// BundleJob tracks the job status for a given bundle
170+
type BundleJob struct {
171+
Name string `json:"name, omitempty"`
172+
Namespace string `json:"namespace, omitempty"`
173+
Condition batchv1.JobConditionType `json:"condition, omitempty"`
174+
CompletionTime *metav1.Time `json:"completionTime, omitempty"`
175+
}
176+
177+
// BundleLookup serves as accounting for tracking a bundle data lookup
178+
type BundleLookup struct {
179+
BundleJob *BundleJob `json:"bundleJob"`
180+
ConfigMapRef *ConfigMapResourceReference `json:"configMapRef"`
181+
Image string `json:"image"`
182+
BundleFromRegistry *api.Bundle `json:"bundleFromRegistry"`
183+
CatalogName string `json:"catalogName"`
184+
CatalogNamespace string `json:"catalogNamespace"`
185+
}
186+
165187
// ManifestsMatch returns true if the CSV manifests in the StepResources of the given list of steps
166188
// matches those in the InstallPlanStatus.
167189
func (s *InstallPlanStatus) CSVManifestsMatch(steps []*Step) bool {

‎pkg/controller/operators/catalog/operator.go

Lines changed: 138 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111
errorwrap "github.com/pkg/errors"
1212
"github.com/sirupsen/logrus"
1313
"google.golang.org/grpc/connectivity"
14+
batchv1 "k8s.io/api/batch/v1"
1415
corev1 "k8s.io/api/core/v1"
1516
rbacv1 "k8s.io/api/rbac/v1"
1617
"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
@@ -31,6 +32,7 @@ import (
3132

3233
"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/grpc"
3334
sharedtime "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/time"
35+
"github.com/operator-framework/operator-registry/pkg/configmap"
3436

3537
"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/operators/reference"
3638
"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/operators/v1alpha1"
@@ -85,6 +87,8 @@ type Operator struct {
8587
catalogSubscriberIndexer map[string]cache.Indexer
8688
clientAttenuator *scoped.ClientAttenuator
8789
serviceAccountQuerier *scoped.UserDefinedServiceAccountQuerier
90+
bundleLoader *configmap.BundleLoader
91+
configmapRegistryImage string
8892
}
8993

9094
type CatalogSourceSyncFunc func(logger *logrus.Entry, in *v1alpha1.CatalogSource) (out *v1alpha1.CatalogSource, continueSync bool, syncError error)
@@ -140,6 +144,8 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
140144
catalogSubscriberIndexer: map[string]cache.Indexer{},
141145
serviceAccountQuerier: scoped.NewUserDefinedServiceAccountQuerier(logger, crClient),
142146
clientAttenuator: scoped.NewClientAttenuator(logger, config, opClient, crClient),
147+
bundleLoader: configmap.NewBundleLoader(),
148+
configmapRegistryImage: configmapRegistryImage,
143149
}
144150
op.sources = grpc.NewSourceStore(logger, 10*time.Second, 10*time.Minute, op.syncSourceState)
145151
op.reconciler = reconciler.NewRegistryReconcilerFactory(lister, opClient, configmapRegistryImage, op.now)
@@ -167,10 +173,13 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
167173
// Wire InstallPlans
168174
ipInformer := crInformerFactory.Operators().V1alpha1().InstallPlans()
169175
op.lister.OperatorsV1alpha1().RegisterInstallPlanLister(namespace, ipInformer.Lister())
176+
ipQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), fmt.Sprintf("%s/ips", namespace))
177+
op.ipQueueSet.Set(namespace, ipQueue)
170178
ipQueueInformer, err := queueinformer.NewQueueInformer(
171179
ctx,
172180
queueinformer.WithMetricsProvider(metrics.NewMetricsInstallPlan(op.client)),
173181
queueinformer.WithLogger(op.logger),
182+
queueinformer.WithQueue(ipQueue),
174183
queueinformer.WithInformer(ipInformer.Informer()),
175184
queueinformer.WithSyncer(queueinformer.LegacySyncHandler(op.syncInstallPlans).ToSyncer()),
176185
)
@@ -772,7 +781,7 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
772781
logger.Debug("resolving subscriptions in namespace")
773782

774783
// resolve a set of steps to apply to a cluster, a set of subscriptions to create/update, and any errors
775-
steps, updatedSubs, err := o.resolver.ResolveSteps(namespace, querier)
784+
steps, bundleLookups, updatedSubs, err := o.resolver.ResolveSteps(namespace, querier)
776785
if err != nil {
777786
return err
778787
}
@@ -790,7 +799,7 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
790799
}
791800
}
792801

793-
installPlanReference, err := o.ensureInstallPlan(logger, namespace, subs, installPlanApproval, steps)
802+
installPlanReference, err := o.ensureInstallPlan(logger, namespace, subs, installPlanApproval, steps, bundleLookups)
794803
if err != nil {
795804
logger.WithError(err).Debug("error ensuring installplan")
796805
return err
@@ -931,8 +940,8 @@ func (o *Operator) updateSubscriptionStatus(namespace string, subs []*v1alpha1.S
931940
return err
932941
}
933942

934-
func (o *Operator) ensureInstallPlan(logger *logrus.Entry, namespace string, subs []*v1alpha1.Subscription, installPlanApproval v1alpha1.Approval, steps []*v1alpha1.Step) (*corev1.ObjectReference, error) {
935-
if len(steps) == 0 {
943+
func (o *Operator) ensureInstallPlan(logger *logrus.Entry, namespace string, subs []*v1alpha1.Subscription, installPlanApproval v1alpha1.Approval, steps []*v1alpha1.Step, bundleLookups []*v1alpha1.BundleLookup) (*corev1.ObjectReference, error) {
944+
if len(steps) == 0 && len(bundleLookups) == 0 {
936945
return nil, nil
937946
}
938947

@@ -978,11 +987,11 @@ func (o *Operator) ensureInstallPlan(logger *logrus.Entry, namespace string, sub
978987
}
979988
logger.Warn("no installplan found with matching manifests, creating new one")
980989

981-
return o.createInstallPlan(namespace, subs, installPlanApproval, steps)
990+
return o.createInstallPlan(namespace, subs, installPlanApproval, steps, bundleLookups)
982991
}
983992

984-
func (o *Operator) createInstallPlan(namespace string, subs []*v1alpha1.Subscription, installPlanApproval v1alpha1.Approval, steps []*v1alpha1.Step) (*corev1.ObjectReference, error) {
985-
if len(steps) == 0 {
993+
func (o *Operator) createInstallPlan(namespace string, subs []*v1alpha1.Subscription, installPlanApproval v1alpha1.Approval, steps []*v1alpha1.Step, bundleLookups []*v1alpha1.BundleLookup) (*corev1.ObjectReference, error) {
994+
if len(steps) == 0 && len(bundleLookups) == 0 {
986995
return nil, nil
987996
}
988997

@@ -1027,6 +1036,7 @@ func (o *Operator) createInstallPlan(namespace string, subs []*v1alpha1.Subscrip
10271036
Phase: phase,
10281037
Plan: steps,
10291038
CatalogSources: catalogSources,
1039+
BundleLookups: bundleLookups,
10301040
}
10311041
res, err = o.client.OperatorsV1alpha1().InstallPlans(namespace).UpdateStatus(res)
10321042
if err != nil {
@@ -1036,6 +1046,108 @@ func (o *Operator) createInstallPlan(namespace string, subs []*v1alpha1.Subscrip
10361046
return reference.GetReference(res)
10371047
}
10381048

1049+
func (o *Operator) checkBundleLookups(plan *v1alpha1.InstallPlan) (bool, error) {
1050+
for _, bundleLookup := range plan.Status.BundleLookups {
1051+
if bundleLookup.BundleJob == nil {
1052+
configmap, job, err := configmap.LaunchBundleImage(o.opClient.KubernetesInterface(), bundleLookup.Image, o.configmapRegistryImage, o.namespace)
1053+
if err != nil {
1054+
return false, err
1055+
}
1056+
logrus.Infof("Launched bundle job for image %v", bundleLookup.Image)
1057+
1058+
bundleLookup.BundleJob = &v1alpha1.BundleJob{
1059+
// job condition and completion time will be filled out later (installplan sync)
1060+
Name: job.GetName(),
1061+
Namespace: job.GetNamespace(),
1062+
}
1063+
bundleLookup.ConfigMapRef = &v1alpha1.ConfigMapResourceReference{
1064+
Name: configmap.GetName(),
1065+
Namespace: configmap.GetNamespace(),
1066+
UID: configmap.GetUID(),
1067+
ResourceVersion: configmap.GetResourceVersion(),
1068+
}
1069+
_, err = o.client.OperatorsV1alpha1().InstallPlans(plan.GetNamespace()).UpdateStatus(plan)
1070+
if err != nil {
1071+
return false, err
1072+
}
1073+
1074+
return false, nil
1075+
}
1076+
1077+
if bundleLookup.BundleJob.CompletionTime != nil {
1078+
// already processed
1079+
continue
1080+
}
1081+
1082+
// TODO: instead of doing a get, should just watch all jobs (add ownerref) and update the installplan from there
1083+
job, err := o.opClient.KubernetesInterface().BatchV1().Jobs(bundleLookup.BundleJob.Namespace).Get(bundleLookup.BundleJob.Name, metav1.GetOptions{})
1084+
if err != nil {
1085+
return false, err
1086+
}
1087+
if len(job.Status.Conditions) == 0 || len(job.Status.Conditions) > 0 && job.Status.Conditions[0].Type != batchv1.JobComplete {
1088+
logrus.Infof("Job '%v' for '%v' not yet completed", job.GetName(), bundleLookup.Image)
1089+
return false, nil
1090+
}
1091+
bundleLookup.BundleJob.Condition = job.Status.Conditions[0].Type
1092+
bundleLookup.BundleJob.CompletionTime = job.Status.CompletionTime
1093+
1094+
configmap, err := o.lister.CoreV1().ConfigMapLister().ConfigMaps(bundleLookup.ConfigMapRef.Namespace).Get(bundleLookup.ConfigMapRef.Name)
1095+
if err != nil {
1096+
return false, err
1097+
}
1098+
1099+
// extract data from configmap and write to install plan
1100+
manifest, err := o.bundleLoader.Load(configmap)
1101+
if err != nil {
1102+
return false, err
1103+
}
1104+
1105+
// combine data from the bundle image into what's already known from the registry
1106+
bundleLookup.BundleFromRegistry.CsvName = manifest.Bundle.Name
1107+
bundleLookup.BundleFromRegistry.PackageName = manifest.Bundle.Package
1108+
bundleLookup.BundleFromRegistry.ChannelName = manifest.Bundle.Channel
1109+
_, jsonCSV, _, err := manifest.Bundle.Serialize()
1110+
if err != nil {
1111+
return false, fmt.Errorf("serialize failed: %s", err.Error())
1112+
}
1113+
bundleLookup.BundleFromRegistry.CsvJson = string(jsonCSV)
1114+
bundleLookup.BundleFromRegistry.Object = []string{string(jsonCSV)}
1115+
for _, item := range manifest.Bundle.Objects {
1116+
bytes, err := item.MarshalJSON()
1117+
if err != nil {
1118+
return false, fmt.Errorf("marshall failed: %v", err)
1119+
}
1120+
bundleLookup.BundleFromRegistry.Object = append(bundleLookup.BundleFromRegistry.Object, string(bytes))
1121+
}
1122+
1123+
var olmCSV v1alpha1.ClusterServiceVersion
1124+
err = json.Unmarshal(jsonCSV, &olmCSV)
1125+
if err != nil {
1126+
return false, fmt.Errorf("csv retrieval failed: %s", err.Error())
1127+
}
1128+
1129+
// TODO: refactor with resolver code (and call the subscription stuff too)
1130+
bundleSteps, err := resolver.NewStepResourceFromBundle(bundleLookup.BundleFromRegistry, plan.GetNamespace(), olmCSV.Spec.Replaces, bundleLookup.CatalogName, bundleLookup.CatalogNamespace)
1131+
if err != nil {
1132+
return false, fmt.Errorf("failed to turn bundle into steps: %s", err.Error())
1133+
}
1134+
1135+
// TODO: could this add duplicate steps?
1136+
for _, s := range bundleSteps {
1137+
plan.Status.Plan = append(plan.Status.Plan, &v1alpha1.Step{
1138+
Resolving: olmCSV.GetName(),
1139+
Resource: s,
1140+
Status: v1alpha1.StepStatusUnknown,
1141+
})
1142+
}
1143+
}
1144+
1145+
if _, err := o.client.OperatorsV1alpha1().InstallPlans(plan.GetNamespace()).UpdateStatus(plan); err != nil {
1146+
return false, err
1147+
}
1148+
return true, nil
1149+
}
1150+
10391151
func (o *Operator) syncInstallPlans(obj interface{}) (syncError error) {
10401152
plan, ok := obj.(*v1alpha1.InstallPlan)
10411153
if !ok {
@@ -1052,11 +1164,29 @@ func (o *Operator) syncInstallPlans(obj interface{}) (syncError error) {
10521164

10531165
logger.Info("syncing")
10541166

1055-
if len(plan.Status.Plan) == 0 {
1167+
if len(plan.Status.Plan) == 0 && len(plan.Status.BundleLookups) == 0 {
10561168
logger.Info("skip processing installplan without status - subscription sync responsible for initial status")
10571169
return
10581170
}
10591171

1172+
// handle bundle data before trying to install
1173+
if len(plan.Status.BundleLookups) != 0 {
1174+
finished, err := o.checkBundleLookups(plan)
1175+
if err != nil {
1176+
syncError = fmt.Errorf("checkBundleLookups failed: %v", err)
1177+
return
1178+
}
1179+
if !finished {
1180+
err := o.ipQueueSet.RequeueAfter(plan.GetNamespace(), plan.GetName(), 5*time.Second)
1181+
if err != nil {
1182+
syncError = err
1183+
return
1184+
}
1185+
o.logger.Debug("install plan not yet populated from bundle image, requeueing")
1186+
return
1187+
}
1188+
}
1189+
10601190
querier := o.serviceAccountQuerier.NamespaceQuerier(plan.GetNamespace())
10611191
reference, err := querier()
10621192
if err != nil {

‎pkg/controller/registry/resolver/evolver.go

Lines changed: 26 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,13 @@ func (e *NamespaceGenerationEvolver) checkForUpdates() error {
6262
if err != nil || bundle == nil {
6363
continue
6464
}
65+
if bundle.BundlePath != "" {
66+
e.gen.AddPendingOperator(LaunchBundleImageInfo{
67+
operatorSourceInfo: op.SourceInfo(),
68+
image: bundle.BundlePath,
69+
bundle: bundle,
70+
})
71+
}
6572

6673
o, err := NewOperatorFromBundle(bundle, op.SourceInfo().StartingCSV, *key)
6774
if err != nil {
@@ -90,6 +97,13 @@ func (e *NamespaceGenerationEvolver) addNewOperators(add map[OperatorSourceInfo]
9097
// TODO: log or collect warnings
9198
return errors.Wrapf(err, "%s not found", s)
9299
}
100+
if bundle.BundlePath != "" {
101+
e.gen.AddPendingOperator(LaunchBundleImageInfo{
102+
operatorSourceInfo: &s,
103+
image: bundle.BundlePath,
104+
bundle: bundle,
105+
})
106+
}
93107

94108
o, err := NewOperatorFromBundle(bundle, s.StartingCSV, *key)
95109
if err != nil {
@@ -115,14 +129,23 @@ func (e *NamespaceGenerationEvolver) queryForRequiredAPIs() error {
115129
e.gen.MarkAPIChecked(*api)
116130

117131
// identify the initialSource
118-
initialSource := CatalogKey{}
132+
var initialSource *OperatorSourceInfo
119133
for _, operator := range e.gen.MissingAPIs()[*api] {
120-
initialSource = operator.SourceInfo().Catalog
134+
initialSource = operator.SourceInfo()
121135
break
122136
}
123137

124138
// attempt to find a bundle that provides that api
125-
if bundle, key, err := e.querier.FindProvider(*api, initialSource); err == nil {
139+
if bundle, key, err := e.querier.FindProvider(*api, initialSource.Catalog); err == nil {
140+
if bundle.BundlePath != "" {
141+
e.gen.AddPendingOperator(LaunchBundleImageInfo{
142+
operatorSourceInfo: initialSource,
143+
image: bundle.BundlePath,
144+
bundle: bundle,
145+
})
146+
return nil
147+
}
148+
126149
// add a bundle that provides the api to the generation
127150
o, err := NewOperatorFromBundle(bundle, "", *key)
128151
if err != nil {

0 commit comments

Comments
 (0)