feedc0de/my-rook-config
1
0
Fork 0
forked from rook/rook
Code Pull Requests Activity

operator: refactor use of tprs for the operator kit

Browse Source
This commit is contained in:
Travis Nielsen
2017-06-28 11:44:51 -07:00
parent a5da34fe4c
commit 5d58bc27cb
16 changed files with 669 additions and 576 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/operator/cluster.go
+52 -95
View File
@@ -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
}
pkg/operator/cluster/cluster.go
+4 -2
View File
@@ -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",
}
)
pkg/operator/cluster/pool.go
+4 -2
View File
@@ -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",
}
pkg/operator/event.go
-110
View File
@@ -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
}
pkg/operator/k8sutil/k8sutil.go
+10 -9
View File
@@ -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
pkg/operator/k8sutil/tpr.go
-138
View File
@@ -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()
}
pkg/operator/kit/client.go
+50
View File
@@ -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
}
pkg/operator/kit/log.go
+25
View File
@@ -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")
pkg/operator/kit/resource.go
+144
View File
@@ -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, "")
}
pkg/operator/kit/retry.go
+56
View File
@@ -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)
}
pkg/operator/k8sutil/timer.go → pkg/operator/kit/timer.go
+14 -19
View File
@@ -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
}
}
pkg/operator/kit/watcher.go
+238
View File
@@ -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 <name>.<group> 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
}
pkg/operator/operator.go
+19 -13
View File
@@ -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)
}
pkg/operator/operator_test.go
+2 -2
View File
@@ -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)
pkg/operator/pool.go
+51 -113
View File
@@ -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
}
}
pkg/operator/tpr.go
-73
View File
@@ -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
}