diff --git a/pkg/operator/cluster.go b/pkg/operator/cluster.go index ca9f159c9..6214ee1be 100644 --- a/pkg/operator/cluster.go +++ b/pkg/operator/cluster.go @@ -20,23 +20,20 @@ package operator import ( "encoding/json" - "errors" "fmt" - "net/http" "time" "sync" "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/operator/cluster" - "github.com/rook/rook/pkg/operator/k8sutil" + "github.com/rook/rook/pkg/operator/kit" rookclient "github.com/rook/rook/pkg/rook/client" kwatch "k8s.io/apimachinery/pkg/watch" ) type clusterManager struct { context *clusterd.Context - watchVersion string devicesInUse bool clusters map[string]*cluster.Cluster tracker *tprTracker @@ -44,7 +41,12 @@ type clusterManager struct { // The initiators that create TPRs specific to specific Rook clusters. // For example, pools, object services, and file services, only make sense in the context of a Rook cluster inclusterInitiators []inclusterInitiator - inclusterMgrs []k8sutil.TPRManager + inclusterMgrs []resourceManager +} + +type clusterEvent struct { + Type kwatch.EventType + Object *cluster.Cluster } func newClusterManager(context *clusterd.Context, inclusterInitiators []inclusterInitiator) *clusterManager { @@ -59,11 +61,11 @@ func newClusterManager(context *clusterd.Context, inclusterInitiators []incluste func (m *clusterManager) Manage() { for { logger.Infof("Managing clusters") - err := m.Load() + watchVersion, err := m.Load() if err != nil { logger.Errorf("failed to load cluster. %+v", err) } else { - if err := m.Watch(); err != nil { + if err := m.watchClusters(watchVersion); err != nil { logger.Errorf("failed to watch clusters. %+v", err) } } @@ -72,13 +74,13 @@ func (m *clusterManager) Manage() { } } -func (m *clusterManager) Load() error { +func (m *clusterManager) Load() (string, error) { // Check if there is an existing cluster to recover logger.Info("finding existing clusters...") clusterList, err := m.getClusterList() if err != nil { - return err + return "", err } logger.Infof("found %d clusters", len(clusterList.Items)) for i := range clusterList.Items { @@ -87,8 +89,7 @@ func (m *clusterManager) Load() error { m.startCluster(&c) } - m.watchVersion = clusterList.Metadata.ResourceVersion - return nil + return clusterList.Metadata.ResourceVersion, nil } func (m *clusterManager) startTrack(c *cluster.Cluster) error { @@ -140,7 +141,7 @@ func (m *clusterManager) startCluster(c *cluster.Cluster) { logger.Infof("starting cluster %s in namespace %s", c.Name, c.Namespace) // Start the Rook cluster components. Retry several times in case of failure. - err := k8sutil.Retry(m.context, func() (bool, error) { + err := k8sutil.Retry(m.context.KubeContext, func() (bool, error) { err := c.CreateInstance() if err != nil { logger.Errorf("failed to create cluster %s in namespace %s. %+v", c.Name, c.Namespace, err) @@ -155,10 +156,10 @@ func (m *clusterManager) startCluster(c *cluster.Cluster) { // Start all the TPRs for this cluster for _, initiator := range m.inclusterInitiators { - k8sutil.Retry(m.context, func() (bool, error) { + k8sutil.Retry(m.context.KubeContext, func() (bool, error) { tprMgr, err := initiator.Create(m, c.Namespace) if err != nil { - logger.Warningf("cannot create in-cluster tpr %s. %+v. retrying...", initiator.Scheme().Name, err) + logger.Warningf("cannot create in-cluster tpr %s. %+v. retrying...", initiator.Resource().Name, err) return false, nil } @@ -192,7 +193,7 @@ func (m *clusterManager) isClustersCacheStale(currentClusters []cluster.Cluster) } func (m *clusterManager) getClusterList() (*cluster.ClusterList, error) { - b, err := k8sutil.GetRawList(m.context.Clientset, cluster.ClusterScheme) + b, err := kit.GetRawList(m.context.Clientset, cluster.ClusterResource) if err != nil { return nil, err } @@ -204,103 +205,46 @@ func (m *clusterManager) getClusterList() (*cluster.ClusterList, error) { return clusters, nil } -func (m *clusterManager) Watch() error { - logger.Infof("start watching cluster tpr: %s", m.watchVersion) +func (m *clusterManager) watchClusters(watchVersion string) error { defer m.tracker.stop() - eventCh, errCh := m.watch() - - go func() { - timer := k8sutil.NewPanicTimer( - time.Minute, - "unexpected long blocking (> 1 Minute) when handling cluster event") - - for event := range eventCh { - timer.Start() - - c := event.Object - - switch event.Type { - case kwatch.Added: - logger.Infof("starting new cluster %s in namespace %s", c.Name, c.Namespace) - m.startCluster(c) - - case kwatch.Modified: - logger.Infof("modifying a cluster not implemented") - - case kwatch.Deleted: - logger.Infof("deleting a cluster not implemented") - } - - timer.Stop() - } - }() - return <-errCh + w := kit.NewWatcher( + m.context.KubeContext, + cluster.ClusterResource, + "", + watchVersion, + m.handleClusterEvent, + m.checkStaleCache) + return w.Watch() } -// watch creates a go routine, and watches the cluster.rook kind resources from -// the given watch version. It emits events on the resources through the returned -// event chan. Errors will be reported through the returned error chan. The go routine -// exits on any error. -func (m *clusterManager) watch() (<-chan *clusterEvent, <-chan error) { - eventCh := make(chan *clusterEvent) - // On unexpected error case, the operator should exit - errCh := make(chan error, 1) +func (m *clusterManager) handleClusterEvent(e *kit.RawEvent) error { - go func() { - defer close(eventCh) - - for { - err := m.watchOuterTPR(eventCh, errCh) - if err != nil { - logger.Warningf("cancelling cluster tpr watch. %+v", err) - return - } - } - }() - - return eventCh, errCh -} - -func (m *clusterManager) watchOuterTPR(eventCh chan *clusterEvent, errCh chan error) error { - resp, err := k8sutil.WatchTPR(m.context, cluster.ClusterScheme, m.watchVersion) + event, err := unmarshalEvent(e) if err != nil { - errCh <- err return err } - defer resp.Body.Close() + switch event.Type { + case kwatch.Added: + logger.Infof("starting new cluster in namespace %s", event.Object.Namespace) + m.startCluster(event.Object) - if resp.StatusCode != http.StatusOK { - err := errors.New("invalid status code: " + resp.Status) - errCh <- err - return err - } + case kwatch.Modified: + logger.Infof("modifying a cluster not implemented") - decoder := json.NewDecoder(resp.Body) - for { - ev, st, err := pollClusterEvent(decoder) - done, err := handlePollEventResult(st, err, m.checkStaleCache, errCh) - if err != nil { - return err - } - if done { - return nil - } - logger.Debugf("rook cluster event: %+v", ev) - - m.watchVersion = ev.Object.ResourceVersion - eventCh <- ev + case kwatch.Deleted: + logger.Infof("deleting a cluster not implemented") } + return nil } -func (m *clusterManager) checkStaleCache() (bool, error) { +func (m *clusterManager) checkStaleCache() (string, error) { clusterList, err := m.getClusterList() if err == nil && !m.isClustersCacheStale(clusterList.Items) { - m.watchVersion = clusterList.Metadata.ResourceVersion - return false, nil + return clusterList.Metadata.ResourceVersion, nil } - return true, err + return "", err } func (m *clusterManager) getRookClient(namespace string) (rookclient.RookRestClient, error) { @@ -322,3 +266,16 @@ func (m *clusterManager) getCluster(namespace string) (*cluster.Cluster, error) return nil, fmt.Errorf("cluster namespace %s not found", namespace) } + +func unmarshalEvent(event *kit.RawEvent) (*clusterEvent, error) { + + e := &clusterEvent{ + Type: event.Type, + Object: &cluster.Cluster{}, + } + err := json.Unmarshal(event.Object, e.Object) + if err != nil { + return nil, fmt.Errorf("fail to unmarshal Cluster object: %v", err) + } + return e, nil +} diff --git a/pkg/operator/cluster/cluster.go b/pkg/operator/cluster/cluster.go index 7b7bcc962..1abda75b1 100644 --- a/pkg/operator/cluster/cluster.go +++ b/pkg/operator/cluster/cluster.go @@ -31,6 +31,7 @@ import ( "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/operator/api" "github.com/rook/rook/pkg/operator/k8sutil" + "github.com/rook/rook/pkg/operator/kit" "github.com/rook/rook/pkg/operator/mgr" "github.com/rook/rook/pkg/operator/mon" "github.com/rook/rook/pkg/operator/osd" @@ -49,9 +50,10 @@ var ( healthCheckInterval = 10 * time.Second clientTimeout = 15 * time.Second logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-cluster") - ClusterScheme = k8sutil.TPRScheme{ + ClusterResource = kit.CustomResource{ Name: "cluster", - Version: k8sutil.V1Alpha1, + Group: k8sutil.CustomResourceGroup, + Version: kit.V1Alpha1, Description: "Managed Rook clusters", } ) diff --git a/pkg/operator/cluster/pool.go b/pkg/operator/cluster/pool.go index 35aa6ecb5..3018d79a7 100644 --- a/pkg/operator/cluster/pool.go +++ b/pkg/operator/cluster/pool.go @@ -22,6 +22,7 @@ import ( "fmt" "github.com/rook/rook/pkg/operator/k8sutil" + "github.com/rook/rook/pkg/operator/kit" rookclient "github.com/rook/rook/pkg/rook/client" "k8s.io/client-go/pkg/api/v1" @@ -33,9 +34,10 @@ const ( erasureCodeType = "erasure-coded" ) -var PoolScheme = k8sutil.TPRScheme{ +var PoolResource = kit.CustomResource{ Name: "pool", - Version: k8sutil.V1Alpha1, + Group: k8sutil.CustomResourceGroup, + Version: kit.V1Alpha1, Description: "Managed Rook pools", } diff --git a/pkg/operator/event.go b/pkg/operator/event.go deleted file mode 100644 index a211f184d..000000000 --- a/pkg/operator/event.go +++ /dev/null @@ -1,110 +0,0 @@ -/* -Copyright 2016 The Rook Authors. All rights reserved. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. - -Some of the code below came from https://github.com/coreos/etcd-operator -which also has the apache 2.0 license. -*/ -package operator - -import ( - "encoding/json" - "fmt" - "io" - - "github.com/rook/rook/pkg/operator/cluster" - - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - kwatch "k8s.io/apimachinery/pkg/watch" -) - -type clusterEvent struct { - Type kwatch.EventType - Object *cluster.Cluster -} - -type poolEvent struct { - Type kwatch.EventType - Object *cluster.Pool -} - -type rawEvent struct { - Type kwatch.EventType - Object json.RawMessage -} - -func pollClusterEvent(decoder *json.Decoder) (*clusterEvent, *metav1.Status, error) { - re, status, err := pollEvent(decoder) - if err != nil { - return nil, nil, fmt.Errorf("failed to poll cluster event. %+v", err) - } - - if status != nil { - return nil, status, nil - } - - ev := &clusterEvent{ - Type: re.Type, - Object: &cluster.Cluster{}, - } - err = json.Unmarshal(re.Object, ev.Object) - if err != nil { - return nil, nil, fmt.Errorf("fail to unmarshal Cluster object from data (%s): %v", re.Object, err) - } - return ev, status, nil -} - -func pollPoolEvent(decoder *json.Decoder) (*poolEvent, *metav1.Status, error) { - re, status, err := pollEvent(decoder) - if err != nil { - return nil, status, fmt.Errorf("failed to poll pool event. %+v", err) - } - - if status != nil { - return nil, status, nil - } - - ev := &poolEvent{ - Type: re.Type, - Object: &cluster.Pool{}, - } - err = json.Unmarshal(re.Object, ev.Object) - if err != nil { - return nil, nil, fmt.Errorf("fail to unmarshal Pool object from data (%s): %v", re.Object, err) - } - return ev, nil, nil -} - -func pollEvent(decoder *json.Decoder) (*rawEvent, *metav1.Status, error) { - re := &rawEvent{} - err := decoder.Decode(re) - if err != nil { - if err == io.EOF { - return nil, nil, err - } - return nil, nil, fmt.Errorf("fail to decode raw event from apiserver (%v)", err) - } - - if re.Type == kwatch.Error { - status := &metav1.Status{} - err = json.Unmarshal(re.Object, status) - if err != nil { - return nil, nil, fmt.Errorf("fail to decode (%s) into metav1.Status (%v)", re.Object, err) - } - logger.Infof("returning pollEvent status %+v", status) - return nil, status, nil - } - - return re, nil, nil -} diff --git a/pkg/operator/k8sutil/k8sutil.go b/pkg/operator/k8sutil/k8sutil.go index e4fb57f68..9041cef68 100644 --- a/pkg/operator/k8sutil/k8sutil.go +++ b/pkg/operator/k8sutil/k8sutil.go @@ -34,12 +34,13 @@ import ( var logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-k8sutil") const ( - Namespace = "rook" - DefaultNamespace = "default" - DataDirVolume = "rook-data" - DataDir = "/var/lib/rook" - RookType = "kubernetes.io/rook" - RbdType = "kubernetes.io/rbd" + Namespace = "rook" + CustomResourceGroup = "rook.io" + DefaultNamespace = "default" + DataDirVolume = "rook-data" + DataDir = "/var/lib/rook" + RookType = "kubernetes.io/rook" + RbdType = "kubernetes.io/rbd" ) type ConditionFunc func() (bool, error) @@ -48,7 +49,7 @@ type ConditionFunc func() (bool, error) // The interval won't be affected by how long f takes. // For example, if interval is 3s, f takes 1s, another f will be called 2s later. // However, if f takes longer than interval, it will be delayed. -func Retry(context *clusterd.Context, f ConditionFunc) error { +func Retry(context clusterd.KubeContext, f ConditionFunc) error { interval := time.Duration(context.RetryDelay) * time.Second tick := time.NewTicker(interval) defer tick.Stop() @@ -68,14 +69,14 @@ func Retry(context *clusterd.Context, f ConditionFunc) error { return fmt.Errorf("failed after max retries %d.", context.MaxRetries) } -func NewHttpClient() (*rest.RESTClient, error) { +func NewHTTPClient() (*rest.RESTClient, error) { config, err := rest.InClusterConfig() if err != nil { return nil, err } config.GroupVersion = &schema.GroupVersion{ - Group: tprGroup, + Group: CustomResourceGroup, } config.APIPath = "/apis" config.ContentType = runtime.ContentTypeJSON diff --git a/pkg/operator/k8sutil/tpr.go b/pkg/operator/k8sutil/tpr.go deleted file mode 100644 index 5afdd1efe..000000000 --- a/pkg/operator/k8sutil/tpr.go +++ /dev/null @@ -1,138 +0,0 @@ -/* -Copyright 2016 The Rook Authors. All rights reserved. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. - -Some of the code was modified from https://github.com/coreos/etcd-operator -which also has the apache 2.0 license. -*/ -package k8sutil - -import ( - "fmt" - "net/http" - - "github.com/rook/rook/pkg/clusterd" - - "k8s.io/apimachinery/pkg/api/errors" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/client-go/kubernetes" - "k8s.io/client-go/pkg/apis/extensions/v1beta1" -) - -const ( - tprGroup = "rook.io" - V1Alpha1 = "v1alpha1" - V1Beta1 = "v1beta1" - V1 = "v1" -) - -type TPRScheme struct { - Name string - Version string - Description string -} - -type TPRManager interface { - Load() error - Watch() error - Manage() -} - -func CreateTPRs(context *clusterd.Context, tprs []TPRScheme) error { - for _, tpr := range tprs { - if err := CreateTPR(context, tpr); err != nil { - return fmt.Errorf("failed to init tpr %s. %+v", tpr.Name, err) - } - } - - for _, tpr := range tprs { - if err := waitForTPRInit(context, tpr); err != nil { - return fmt.Errorf("failed to complete init %s. %+v", tpr.Name, err) - } - } - - return nil -} - -func CreateTPR(context *clusterd.Context, tpr TPRScheme) error { - logger.Infof("creating %s TPR", tpr.Name) - r := &v1beta1.ThirdPartyResource{ - ObjectMeta: metav1.ObjectMeta{ - Name: fmt.Sprintf("%s.%s", tpr.Name, tprGroup), - }, - Versions: []v1beta1.APIVersion{ - {Name: tpr.Version}, - }, - Description: tpr.Description, - } - _, err := context.Clientset.ExtensionsV1beta1().ThirdPartyResources().Create(r) - if err != nil { - if !errors.IsAlreadyExists(err) { - return fmt.Errorf("failed to create %s TPR. %+v", tpr.Name, err) - } - } - - return nil -} - -func waitForTPRInit(context *clusterd.Context, scheme TPRScheme) error { - restcli := context.Clientset.CoreV1().RESTClient() - uri := TPRURI(scheme) - return Retry(context, func() (bool, error) { - _, err := restcli.Get().RequestURI(uri).DoRaw() - if err != nil { - logger.Infof("did not yet find tpr %s at %s. %+v", scheme.Name, uri, err) - if errors.IsNotFound(err) { - return false, nil - } - return false, err - } - return true, nil - }) -} - -func WatchTPRNamespaced(context *clusterd.Context, scheme TPRScheme, namespace, resourceVersion string) (*http.Response, error) { - uri := fmt.Sprintf("%s/%s?watch=true&resourceVersion=%s", context.MasterHost, TPRURINamespaced(scheme, namespace), resourceVersion) - logger.Debugf("watching tpr: %s", uri) - return context.KubeHttpCli.Get(uri) -} - -func WatchTPR(context *clusterd.Context, scheme TPRScheme, resourceVersion string) (*http.Response, error) { - uri := fmt.Sprintf("%s/%s?watch=true&resourceVersion=%s", context.MasterHost, TPRURI(scheme), resourceVersion) - logger.Debugf("watching tpr: %s", uri) - return context.KubeHttpCli.Get(uri) -} - -func TPRURINamespaced(scheme TPRScheme, namespace string) string { - // /apis/rook.io/v1alpha1/namespaces/rook/pools - return fmt.Sprintf("apis/%s/%s/namespaces/%s/%ss", tprGroup, scheme.Version, namespace, scheme.Name) -} - -func TPRURI(scheme TPRScheme) string { - // creates a uri that is for retrieving or watching a tpr in all namespaces. For example: - // /apis/rook.io/v1alpha1/clusters - return fmt.Sprintf("apis/%s/%s/%ss", tprGroup, scheme.Version, scheme.Name) -} - -func GetRawListNamespaced(clientset kubernetes.Interface, scheme TPRScheme, namespace string) ([]byte, error) { - restcli := clientset.CoreV1().RESTClient() - uri := TPRURINamespaced(scheme, namespace) - logger.Debugf("getting tpr: %s", uri) - return restcli.Get().RequestURI(uri).DoRaw() -} - -func GetRawList(clientset kubernetes.Interface, scheme TPRScheme) ([]byte, error) { - restcli := clientset.CoreV1().RESTClient() - return restcli.Get().RequestURI(TPRURI(scheme)).DoRaw() -} diff --git a/pkg/operator/kit/client.go b/pkg/operator/kit/client.go new file mode 100644 index 000000000..18828cdef --- /dev/null +++ b/pkg/operator/kit/client.go @@ -0,0 +1,50 @@ +/* +Package kit for Kubernetes operators + +Copyright 2016 The Rook Authors. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +Some of the code below came from https://github.com/coreos/etcd-operator +which also has the apache 2.0 license. +*/ +package kit + +import ( + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/runtime/serializer" + "k8s.io/client-go/pkg/api" + "k8s.io/client-go/rest" +) + +// NewHTTPClient creates a new http client for the operator to manage the Kubernetes cluster +func NewHTTPClient(group string) (*rest.RESTClient, error) { + config, err := rest.InClusterConfig() + if err != nil { + return nil, err + } + + config.GroupVersion = &schema.GroupVersion{ + Group: group, + } + config.APIPath = "/apis" + config.ContentType = runtime.ContentTypeJSON + config.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: api.Codecs} + + restcli, err := rest.RESTClientFor(config) + if err != nil { + return nil, err + } + return restcli, nil +} diff --git a/pkg/operator/kit/log.go b/pkg/operator/kit/log.go new file mode 100644 index 000000000..815816685 --- /dev/null +++ b/pkg/operator/kit/log.go @@ -0,0 +1,25 @@ +/* +Package kit for Kubernetes operators + +Copyright 2016 The Rook Authors. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +Some of the code below came from https://github.com/coreos/etcd-operator +which also has the apache 2.0 license. +*/ +package kit + +import "github.com/coreos/pkg/capnslog" + +var logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-kit") diff --git a/pkg/operator/kit/resource.go b/pkg/operator/kit/resource.go new file mode 100644 index 000000000..3aa36f3c7 --- /dev/null +++ b/pkg/operator/kit/resource.go @@ -0,0 +1,144 @@ +/* +Package kit for Kubernetes operators + +Copyright 2016 The Rook Authors. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +Some of the code was modified from https://github.com/coreos/etcd-operator +which also has the apache 2.0 license. +*/ +package kit + +import ( + "fmt" + "net/http" + + "github.com/rook/rook/pkg/clusterd" + + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/pkg/apis/extensions/v1beta1" +) + +const ( + // V1Alpha1 version for kubernetes resources + V1Alpha1 = "v1alpha1" + + // V1Beta1 version for kubernetes resources + V1Beta1 = "v1beta1" + + // V1 version for kubernetes resources + V1 = "v1" +) + +// CustomResource is for creating a Kubernetes TPR/CRD +type CustomResource struct { + // Name of the custom resource + Name string + + // Group the custom resource belongs to + Group string + + // Version which should be defined in a const above + Version string + + // Description that is human readable + Description string +} + +// CreateCustomResources creates the given custom resources and waits for them to initialize +func CreateCustomResources(context clusterd.KubeContext, resources []CustomResource) error { + for _, resource := range resources { + if err := CreateCustomResource(context, resource); err != nil { + return fmt.Errorf("failed to init resource %s. %+v", resource.Name, err) + } + } + + for _, resource := range resources { + if err := waitForCustomResourceInit(context, resource); err != nil { + return fmt.Errorf("failed to complete init %s. %+v", resource.Name, err) + } + } + + return nil +} + +// CreateCustomResource creates a single custom resource, but does not wait for it to initialize +func CreateCustomResource(context clusterd.KubeContext, resource CustomResource) error { + logger.Infof("creating %s resource", resource.Name) + r := &v1beta1.ThirdPartyResource{ + ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf("%s.%s", resource.Name, resource.Group), + }, + Versions: []v1beta1.APIVersion{ + {Name: resource.Version}, + }, + Description: resource.Description, + } + _, err := context.Clientset.ExtensionsV1beta1().ThirdPartyResources().Create(r) + if err != nil { + if !errors.IsAlreadyExists(err) { + return fmt.Errorf("failed to create %s resource. %+v", resource.Name, err) + } + } + + return nil +} + +func waitForCustomResourceInit(context clusterd.KubeContext, scheme CustomResource) error { + restcli := context.Clientset.CoreV1().RESTClient() + uri := resourceURI(resource, "") + return Retry(context, func() (bool, error) { + _, err := restcli.Get().RequestURI(uri).DoRaw() + if err != nil { + logger.Infof("did not yet find resource %s at %s. %+v", resource.Name, uri, err) + if errors.IsNotFound(err) { + return false, nil + } + return false, err + } + return true, nil + }) +} + +func watchResource(context clusterd.KubeContext, scheme CustomResource, namespace, resourceVersion string) (*http.Response, error) { + uri := fmt.Sprintf("%s/%s?watch=true&resourceVersion=%s", context.MasterHost, resourceURI(scheme, namespace), resourceVersion) + logger.Debugf("watching resource: %s", uri) + return context.KubeHttpCli.Get(uri) +} + +func resourceURI(resource CustomResource, namespace string) string { + if namespace == "" { + // creates a uri that is for retrieving or watching a resource in all namespaces. For example: + // /apis/rook.io/v1alpha1/clusters + return fmt.Sprintf("apis/%s/%s/%ss", resource.Group, resource.Version, resource.Name) + } + + // create a uri that is for a specific namespace + // /apis/rook.io/v1alpha1/namespaces/rook/pools + return fmt.Sprintf("apis/%s/%s/namespaces/%s/%ss", resource.Group, resource.Version, namespace, resource.Name) +} + +// GetRawListNamespaced retrieves a list custom resources of the given type in a specific namespace +func GetRawListNamespaced(clientset kubernetes.Interface, resource CustomResource, namespace string) ([]byte, error) { + restcli := clientset.CoreV1().RESTClient() + uri := resourceURI(resource, namespace) + return restcli.Get().RequestURI(uri).DoRaw() +} + +// GetRawList retrieves a list of custom resources of the given type across all namespaces +func GetRawList(clientset kubernetes.Interface, resource CustomResource) ([]byte, error) { + return GetRawListNamespaced(clientset, resource, "") +} diff --git a/pkg/operator/kit/retry.go b/pkg/operator/kit/retry.go new file mode 100644 index 000000000..254ccaf22 --- /dev/null +++ b/pkg/operator/kit/retry.go @@ -0,0 +1,56 @@ +/* +Package kit for Kubernetes operators + +Copyright 2016 The Rook Authors. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +Some of the code below came from https://github.com/coreos/etcd-operator +which also has the apache 2.0 license. +*/ +package kit + +import ( + "fmt" + "time" + + "github.com/rook/rook/pkg/clusterd" +) + +// ConditionFunc returns true if a retry condition has been satisfied. +// If the condition returns false, the retry will try again. +type ConditionFunc func() (bool, error) + +// Retry retries f every interval until after maxRetries. +// The interval won't be affected by how long f takes. +// For example, if interval is 3s, f takes 1s, another f will be called 2s later. +// However, if f takes longer than interval, it will be delayed. +func Retry(context clusterd.KubeContext, f ConditionFunc) error { + interval := time.Duration(context.RetryDelay) * time.Second + tick := time.NewTicker(interval) + defer tick.Stop() + + for i := 0; i < context.MaxRetries; i++ { + ok, err := f() + if err != nil { + return fmt.Errorf("failed on retry %d. %+v", i, err) + } + if ok { + return nil + } + if i < context.MaxRetries-1 { + <-tick.C + } + } + return fmt.Errorf("failed after max retries %d", context.MaxRetries) +} diff --git a/pkg/operator/k8sutil/timer.go b/pkg/operator/kit/timer.go similarity index 69% rename from pkg/operator/k8sutil/timer.go rename to pkg/operator/kit/timer.go index fefd7244d..012ddcf5a 100644 --- a/pkg/operator/k8sutil/timer.go +++ b/pkg/operator/kit/timer.go @@ -1,4 +1,6 @@ /* +Package kit for Kubernetes operators + Copyright 2016 The Rook Authors. All rights reserved. Licensed under the Apache License, Version 2.0 (the "License"); @@ -16,34 +18,27 @@ limitations under the License. Some of the code below came from https://github.com/coreos/etcd-operator which also has the apache 2.0 license. */ -package k8sutil +package kit import "time" // panicTimer panics when it reaches the given duration. -type PanicTimer struct { - d time.Duration - msg string - t *time.Timer +type panicTimer struct { + duration time.Duration + message string + timer *time.Timer } -func NewPanicTimer(d time.Duration, msg string) *PanicTimer { - return &PanicTimer{ - d: d, - msg: msg, - } -} - -func (pt *PanicTimer) Start() { - pt.t = time.AfterFunc(pt.d, func() { - panic(pt.msg) +func (t *panicTimer) Start() { + t.timer = time.AfterFunc(t.duration, func() { + panic(t.message) }) } // stop stops the timer and resets the elapsed duration. -func (pt *PanicTimer) Stop() { - if pt.t != nil { - pt.t.Stop() - pt.t = nil +func (t *panicTimer) Stop() { + if t.timer != nil { + t.timer.Stop() + t.timer = nil } } diff --git a/pkg/operator/kit/watcher.go b/pkg/operator/kit/watcher.go new file mode 100644 index 000000000..6f8f9ec05 --- /dev/null +++ b/pkg/operator/kit/watcher.go @@ -0,0 +1,238 @@ +/* +Package kit for Kubernetes operators + +Copyright 2016 The Rook Authors. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +Some of the code was modified from https://github.com/coreos/etcd-operator +which also has the apache 2.0 license. +*/ +package kit + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "time" + + "github.com/rook/rook/pkg/clusterd" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + kwatch "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/pkg/api/v1" +) + +var ( + // ErrVersionOutdated indicates that the custom resource is outdated and needs to be refreshed + ErrVersionOutdated = errors.New("requested version is outdated in apiserver") +) + +// ResourceWatcher watches a custom resource for desired state +type ResourceWatcher struct { + context clusterd.KubeContext + resource CustomResource + namespace string + group string + watchVersion string + callback func(event *RawEvent) error + checkStaleCache func() (string, error) +} + +// RawEvent is the raw json message retrieved from the TPR/CRD update +type RawEvent struct { + Type kwatch.EventType + Object json.RawMessage +} + +// NewWatcher creates an instance of a custom resource watcher for the given resource +func NewWatcher(context clusterd.KubeContext, resource CustomResource, namespace, watchVersion string, + callback func(event *RawEvent) error, + checkStaleCache func() (string, error)) *ResourceWatcher { + + // Assign a default func for resources that don't have a cache + if checkStaleCache == nil { + checkStaleCache = func() (string, error) { + // there is no cache for the resource so we indicate to always reset the watch on error + return "", nil + } + } + return &ResourceWatcher{ + context: context, + resource: resource, + namespace: namespace, + watchVersion: watchVersion, + callback: callback, + checkStaleCache: checkStaleCache, + } +} + +// Watch begins watching the custom resource (TPR/CRD). The call will block until an error is raised during the watch. +// When the watch has detected a create, update, or delete event, the raw event will be passed to the caller +// in the callback. After the callback returns, the watch loop will continue for the next event. +// If the callback returns an error, the error will be logged but will not abort the event loop. +func (w *ResourceWatcher) Watch() error { + if w.namespace == "" { + logger.Infof("start watching %s resource in all namespaces at %s", w.resource.Name, w.watchVersion) + } else { + logger.Infof("start watching %s resource in namespace %s at %s", w.resource.Name, w.namespace, w.watchVersion) + } + + eventCh, errCh := w.watch() + + go func() { + + timer := &panicTimer{ + duration: time.Minute, + message: fmt.Sprintf("unexpected long blocking (> 1 Minute) when handling %s event", w.resource.Name), + } + + for event := range eventCh { + timer.Start() + + err := w.callback(event) + if err != nil { + logger.Errorf("failed to handle event %s on %s. %+v", event.Type, w.resource.Name, err) + } + + timer.Stop() + } + }() + return <-errCh +} + +// watch creates a go routine, and watches the custom resource at . starting at +// the given watch version. It emits events on the resources through the returned +// event chan. Errors will be reported through the returned error chan. The go routine +// exits on any error. +func (w *ResourceWatcher) watch() (<-chan *RawEvent, <-chan error) { + eventCh := make(chan *RawEvent) + // On unexpected error case, the operator should exit + errCh := make(chan error, 1) + + go func() { + defer close(eventCh) + + for { + err := w.watchOuterResource(eventCh, errCh) + if err != nil { + errCh <- fmt.Errorf("failed to watch %s resource. %+v", w.resource.Name, err) + return + } + } + }() + + return eventCh, errCh +} + +func (w *ResourceWatcher) watchOuterResource(eventCh chan *RawEvent, errCh chan error) error { + resp, err := watchResource(w.context, w.resource, w.namespace, w.watchVersion) + if err != nil { + errCh <- err + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("invalid status code: %s", resp.Status) + } + + decoder := json.NewDecoder(resp.Body) + for { + ev, st, err := pollEvent(decoder) + done, err := w.handlePollEventResult(st, err, errCh) + if err != nil { + return err + } + if done { + return nil + } + logger.Debugf("rook pool event: %+v", ev) + + // Extract the resource version from the raw json + meta := &v1.ObjectMeta{} + err = json.Unmarshal(ev.Object, meta) + if err != nil { + return fmt.Errorf("fail to unmarshal metadata from body %s: %v", resp.Body, err) + } + w.watchVersion = meta.ResourceVersion + eventCh <- ev + } +} + +func (w *ResourceWatcher) handlePollEventResult(status *metav1.Status, errIn error, errCh chan error) (done bool, err error) { + if errIn != nil { + if errIn == io.EOF { // apiserver will close stream periodically + logger.Debug("apiserver closed stream") + done = true + return + } + + err = errIn + errCh <- fmt.Errorf("received invalid event from API server: %v", err) + return + } + + if status != nil { + + if status.Code == http.StatusGone { + // event history is outdated. + // if nothing has changed, we can go back to watch again. + var resourceVersion string + resourceVersion, err = w.checkStaleCache() + if err == nil && resourceVersion != "" { + // we were able to recover from the cache, so we update the watch version to the latest + // resource version from the cache + w.watchVersion = resourceVersion + done = true + return + } + + // if anything has changed (or error on relist), we have to rebuild the state. + // go to recovery path + err = ErrVersionOutdated + errCh <- ErrVersionOutdated + return + } + + logger.Errorf("unexpected status response from API server: %v", status.Message) + done = true + return + } + return +} + +func pollEvent(decoder *json.Decoder) (*RawEvent, *metav1.Status, error) { + re := &RawEvent{} + err := decoder.Decode(re) + if err != nil { + if err == io.EOF { + return nil, nil, err + } + return nil, nil, fmt.Errorf("fail to decode raw event from apiserver (%v)", err) + } + + if re.Type == kwatch.Error { + status := &metav1.Status{} + err = json.Unmarshal(re.Object, status) + if err != nil { + return nil, nil, fmt.Errorf("fail to decode (%s) into metav1.Status (%v)", re.Object, err) + } + logger.Infof("returning pollEvent status %+v", status) + return nil, status, nil + } + + return re, nil, nil +} diff --git a/pkg/operator/operator.go b/pkg/operator/operator.go index 58ce3130d..e93f86dbf 100644 --- a/pkg/operator/operator.go +++ b/pkg/operator/operator.go @@ -19,7 +19,6 @@ which also has the apache 2.0 license. package operator import ( - "errors" "fmt" "time" @@ -27,6 +26,7 @@ import ( "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/operator/cluster" "github.com/rook/rook/pkg/operator/k8sutil" + "github.com/rook/rook/pkg/operator/kit" "k8s.io/apimachinery/pkg/util/wait" ) @@ -46,30 +46,36 @@ const ( termLimit = controller.DefaultTermLimit ) -var ( - ErrVersionOutdated = errors.New("requested version is outdated in apiserver") -) - type Operator struct { - context *clusterd.Context - tprSchemes []k8sutil.TPRScheme - // The TPR that is global to the kubernetes cluster. - // The cluster TPR is global because you create multiple clusers in k8s + context *clusterd.Context + resources []kit.CustomResource + // The custom resource that is global to the kubernetes cluster. + // The cluster is global because you create multiple clusers in k8s clusterMgr *clusterManager volumeProvisioner controller.Provisioner } +type inclusterInitiator interface { + Create(clusterMgr *clusterManager, namespace string) (resourceManager, error) + Resource() kit.CustomResource +} + +type resourceManager interface { + Load() (string, error) + Manage() +} + func New(context *clusterd.Context) *Operator { poolInitiator := newPoolInitiator(context) clusterMgr := newClusterManager(context, []inclusterInitiator{poolInitiator}) volumeProvisioner := newRookVolumeProvisioner(clusterMgr) - schemes := []k8sutil.TPRScheme{cluster.ClusterScheme, cluster.PoolScheme} + schemes := []kit.CustomResource{cluster.ClusterResource, cluster.PoolResource} return &Operator{ context: context, clusterMgr: clusterMgr, - tprSchemes: schemes, + resources: schemes, volumeProvisioner: volumeProvisioner, } } @@ -112,13 +118,13 @@ func (o *Operator) Run() error { } func (o *Operator) initResources() error { - httpCli, err := k8sutil.NewHttpClient() + httpCli, err := k8sutil.NewHTTPClient() if err != nil { return fmt.Errorf("failed to get tpr client. %+v", err) } o.context.KubeHttpCli = httpCli.Client - err = k8sutil.CreateTPRs(o.context, o.tprSchemes) + err = kit.CreateCustomResources(o.context.KubeContext, o.resources) if err != nil { return fmt.Errorf("failed to create TPR. %+v", err) } diff --git a/pkg/operator/operator_test.go b/pkg/operator/operator_test.go index 4b1e2205b..c24de09f2 100644 --- a/pkg/operator/operator_test.go +++ b/pkg/operator/operator_test.go @@ -36,8 +36,8 @@ func TestCreateCluster(t *testing.T) { assert.NotNil(t, err) // create the tpr - scheme := k8sutil.TPRScheme{Name: "test", Version: "alpha", Description: "test description"} - err = k8sutil.CreateTPR(o.context, scheme) + scheme := k8sutil.CustomResource{Name: "test", Version: "alpha", Description: "test description"} + err = k8sutil.CreateCustomResource(o.context, scheme) assert.Nil(t, err) tpr, err := clientset.ExtensionsV1beta1().ThirdPartyResources().List(v1.ListOptions{}) assert.Nil(t, err) diff --git a/pkg/operator/pool.go b/pkg/operator/pool.go index 8a264a83b..af0e290cb 100644 --- a/pkg/operator/pool.go +++ b/pkg/operator/pool.go @@ -20,14 +20,12 @@ package operator import ( "encoding/json" - "errors" "fmt" - "net/http" "time" "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/operator/cluster" - "github.com/rook/rook/pkg/operator/k8sutil" + "github.com/rook/rook/pkg/operator/kit" rookclient "github.com/rook/rook/pkg/rook/client" kwatch "k8s.io/apimachinery/pkg/watch" ) @@ -43,11 +41,16 @@ type poolManager struct { rclient rookclient.RookRestClient } +type poolEvent struct { + Type kwatch.EventType + Object *cluster.Pool +} + func newPoolInitiator(context *clusterd.Context) *poolInitiator { return &poolInitiator{context: context} } -func (p *poolInitiator) Create(clusterMgr *clusterManager, namespace string) (k8sutil.TPRManager, error) { +func (p *poolInitiator) Create(clusterMgr *clusterManager, namespace string) (resourceManager, error) { rclient, err := clusterMgr.getRookClient(namespace) if err != nil { return nil, fmt.Errorf("failed to get api client for pool tpr for cluster in namespace %s. %+v", namespace, err) @@ -55,8 +58,8 @@ func (p *poolInitiator) Create(clusterMgr *clusterManager, namespace string) (k8 return &poolManager{context: p.context, namespace: namespace, rclient: rclient}, nil } -func (p *poolInitiator) Scheme() k8sutil.TPRScheme { - return cluster.PoolScheme +func (p *poolInitiator) Resource() kit.CustomResource { + return cluster.PoolResource } // Run the tpr manager until the caller signals with an EndWatch() @@ -64,11 +67,13 @@ func (p *poolManager) Manage() { for { // load and initialize the pools - if err := p.Load(); err != nil { + watchVersion, err := p.Load() + if err != nil { logger.Errorf("cannot load %s pool tpr. %+v. retrying...", p.namespace, err) } else { // watch for added/updated/deleted pools - if err := p.Watch(); err != nil { + watcher := kit.NewWatcher(p.context.KubeContext, cluster.PoolResource, p.namespace, watchVersion, p.handlePoolEvent, nil) + if err := watcher.Watch(); err != nil { logger.Errorf("failed to watch %s pool tpr. %+v. retrying...", p.namespace, err) } } @@ -77,12 +82,46 @@ func (p *poolManager) Manage() { } } -func (p *poolManager) Load() error { +func (p *poolManager) handlePoolEvent(event *kit.RawEvent) error { + pool := &poolEvent{ + Type: event.Type, + Object: &cluster.Pool{}, + } + err := json.Unmarshal(event.Object, pool.Object) + if err != nil { + return fmt.Errorf("fail to unmarshal Pool from data (%s): %v", pool.Object, err) + } + + switch event.Type { + case kwatch.Added: + if err := pool.Object.Create(p.rclient); err != nil { + logger.Errorf("failed to create pool %s. %+v", pool.Object.Name, err) + break + } + + case kwatch.Modified: + // if the pool is modified, allow the pool to be created if it wasn't already + if err := pool.Object.Create(p.rclient); err != nil { + logger.Errorf("failed to create (modify) pool %s. %+v", pool.Object.Name, err) + break + } + + case kwatch.Deleted: + err := pool.Object.Delete(p.rclient) + if err != nil { + logger.Errorf("failed to delete pool %s. %+v", pool.Object.Name, err) + break + } + } + return nil +} + +func (p *poolManager) Load() (string, error) { // Check if pools have all been created logger.Info("finding existing pools...") poolList, err := p.getPoolList() if err != nil { - return err + return "", err } logger.Infof("found %d pools. ensuring they exist.", len(poolList.Items)) @@ -95,12 +134,11 @@ func (p *poolManager) Load() error { } } - p.watchVersion = poolList.Metadata.ResourceVersion - return nil + return poolList.Metadata.ResourceVersion, nil } func (p *poolManager) getPoolList() (*cluster.PoolList, error) { - b, err := k8sutil.GetRawListNamespaced(p.context.Clientset, cluster.PoolScheme, p.namespace) + b, err := kit.GetRawListNamespaced(p.context.Clientset, cluster.PoolResource, p.namespace) if err != nil { return nil, err } @@ -111,103 +149,3 @@ func (p *poolManager) getPoolList() (*cluster.PoolList, error) { } return pools, nil } - -func (p *poolManager) Watch() error { - logger.Infof("start watching pool tpr in namespace %s: %s", p.namespace, p.watchVersion) - - eventCh, errCh := p.watch() - - go func() { - timer := k8sutil.NewPanicTimer( - time.Minute, - "unexpected long blocking (> 1 Minute) when handling pool event") - - for event := range eventCh { - timer.Start() - - pool := event.Object - - switch event.Type { - case kwatch.Added: - if err := pool.Create(p.rclient); err != nil { - logger.Errorf("failed to create pool %s. %+v", pool.Name, err) - break - } - - case kwatch.Modified: - // if the pool is modified, allow the pool to be created if it wasn't already - if err := pool.Create(p.rclient); err != nil { - logger.Errorf("failed to create (modify) pool %s. %+v", pool.Name, err) - break - } - - case kwatch.Deleted: - err := pool.Delete(p.rclient) - if err != nil { - logger.Errorf("failed to delete pool %s. %+v", pool.Name, err) - break - } - } - - timer.Stop() - } - }() - return <-errCh - -} - -// watch creates a go routine, and watches the pool.rook.io kind resources from -// the given watch version. It emits events on the resources through the returned -// event chan. Errors will be reported through the returned error chan. The go routine -// exits on any error. -func (p *poolManager) watch() (<-chan *poolEvent, <-chan error) { - eventCh := make(chan *poolEvent) - // On unexpected error case, the operator should exit - errCh := make(chan error, 1) - - go func() { - defer close(eventCh) - - for { - err := p.watchOuterTPR(eventCh, errCh) - if err != nil { - errCh <- fmt.Errorf("failed to watch pool tpr. %+v", err) - return - } - } - }() - - return eventCh, errCh -} - -func (p *poolManager) watchOuterTPR(eventCh chan *poolEvent, errCh chan error) error { - resp, err := k8sutil.WatchTPRNamespaced(p.context, cluster.PoolScheme, p.namespace, p.watchVersion) - if err != nil { - errCh <- err - return err - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return errors.New("invalid status code: " + resp.Status) - } - - decoder := json.NewDecoder(resp.Body) - for { - ev, st, err := pollPoolEvent(decoder) - done, err := handlePollEventResult(st, err, func() (bool, error) { - // there is no cache for the pool tpr, so we return true to indicate to always reset the watch on error - return true, nil - }, errCh) - if err != nil { - return err - } - if done { - return nil - } - logger.Debugf("rook pool event: %+v", ev) - - p.watchVersion = ev.Object.ResourceVersion - eventCh <- ev - } -} diff --git a/pkg/operator/tpr.go b/pkg/operator/tpr.go deleted file mode 100644 index 0fbea612e..000000000 --- a/pkg/operator/tpr.go +++ /dev/null @@ -1,73 +0,0 @@ -/* -Copyright 2016 The Rook Authors. All rights reserved. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. - -Some of the code was modified from https://github.com/coreos/etcd-operator -which also has the apache 2.0 license. -*/ -package operator - -import ( - "fmt" - "io" - "net/http" - - "github.com/rook/rook/pkg/operator/k8sutil" - - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" -) - -type inclusterInitiator interface { - Create(clusterMgr *clusterManager, namespace string) (k8sutil.TPRManager, error) - Scheme() k8sutil.TPRScheme -} - -func handlePollEventResult(status *metav1.Status, errIn error, checkStaleCache func() (bool, error), errCh chan error) (done bool, err error) { - if errIn != nil { - if errIn == io.EOF { // apiserver will close stream periodically - logger.Debug("apiserver closed stream") - done = true - return - } - - err = errIn - errCh <- fmt.Errorf("received invalid event from API server: %v", err) - return - } - - if status != nil { - - if status.Code == http.StatusGone { - // event history is outdated. - // if nothing has changed, we can go back to watch again. - var stale bool - stale, err = checkStaleCache() - if err == nil && !stale { - done = true - return - } - - // if anything has changed (or error on relist), we have to rebuild the state. - // go to recovery path - err = ErrVersionOutdated - errCh <- ErrVersionOutdated - return - } - - logger.Errorf("unexpected status response from API server: %v", status.Message) - done = true - return - } - return -}