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

watch for pool tprs inside the cluster namespace

Browse Source
This commit is contained in:
Travis Nielsen
2017-04-11 12:01:52 -07:00
parent ba79229fb8
commit d685481c10
8 changed files with 182 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
demo/kubernetes/rook-pool.yaml
+1 -2
View File
@@ -1,10 +1,9 @@
apiVersion: rook.io/v1alpha1
kind: Rookpool
metadata:
name: rook-replicapool
spec:
name: replicapool
namespace: rook
spec:
replication:
count: 1
# For an erasure-coded pool, comment out the replication count above and uncomment the following settings.
pkg/operator/cluster.go
+56 -29
View File
@@ -33,42 +33,39 @@ import (
kwatch "k8s.io/client-go/pkg/watch"
)
type clusterTPR struct {
type clusterManager struct {
context *context
watchVersion string
devicesInUse bool
clusters map[string]*cluster.Cluster
tracker *tprTracker
sync.RWMutex
// 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 []tprManager
}
func newClusterTPR(context *context) *clusterTPR {
return &clusterTPR{
context: context,
clusters: make(map[string]*cluster.Cluster),
tracker: newTPRTracker(),
func newClusterManager(context *context, inclusterInitiators []inclusterInitiator) *clusterManager {
return &clusterManager{
context: context,
clusters: make(map[string]*cluster.Cluster),
tracker: newTPRTracker(),
inclusterInitiators: inclusterInitiators,
}
}
func (t *clusterTPR) Name() string {
func (t *clusterManager) Name() string {
return "rookcluster"
}
func (t *clusterTPR) Description() string {
func (t *clusterManager) Description() string {
return "Managed Rook clusters"
}
func (t *clusterTPR) Load() error {
func (t *clusterManager) Load() error {
// Check if there is an existing cluster to recover
err := t.findAllClusters()
if err != nil {
return fmt.Errorf("failed to find clusters. %+v", err)
}
return nil
}
func (t *clusterTPR) findAllClusters() error {
logger.Info("finding existing clusters...")
clusterList, err := t.getClusterList()
if err != nil {
@@ -91,7 +88,7 @@ func (t *clusterTPR) findAllClusters() error {
return nil
}
func (t *clusterTPR) startCluster(c *cluster.Cluster) {
func (t *clusterManager) startCluster(c *cluster.Cluster) {
if t.devicesInUse && c.Spec.Storage.AnyUseAllDevices() {
logger.Warningf("devices in more than one namespace not supported. ignoring devices in namespace %s", c.Spec.Namespace)
c.Spec.Storage.ClearUseAllDevices()
@@ -102,16 +99,41 @@ func (t *clusterTPR) startCluster(c *cluster.Cluster) {
}
go func() {
// Start the Rook cluster components
err := c.CreateInstance()
if err != nil {
logger.Errorf("failed to create cluster in namespace %s. %+v", c.Spec.Namespace, err)
return
}
// Start all the TPRs for this cluster
for _, tpr := range t.inclusterInitiators {
var tprMgr tprManager
k8sutil.Retry(time.Duration(t.context.retryDelay)*time.Second, t.context.maxRetries, func() (bool, error) {
if tprMgr, err = tpr.Create(t, c.Namespace); err != nil {
logger.Warningf("cannot create %s in-cluster tpr %s. %+v. retrying...", t.Name(), err)
return false, nil
}
if err := tprMgr.Load(); err != nil {
logger.Warningf("cannot load %s in-cluster tpr %s. %+v. retrying...", t.Name(), err)
return false, nil
}
// Start the tpr-manager asynchronously
t.Lock()
defer t.Unlock()
t.inclusterMgrs = append(t.inclusterMgrs, tprMgr)
go tprMgr.BeginWatch()
return true, nil
})
}
c.Monitor(t.tracker.stopChMap[c.Spec.Namespace])
}()
}
func (t *clusterTPR) isClustersCacheStale(currentClusters []cluster.Cluster) bool {
func (t *clusterManager) isClustersCacheStale(currentClusters []cluster.Cluster) bool {
if len(t.tracker.clusterRVs) != len(currentClusters) {
return true
}
@@ -126,8 +148,8 @@ func (t *clusterTPR) isClustersCacheStale(currentClusters []cluster.Cluster) boo
return false
}
func (t *clusterTPR) getClusterList() (*cluster.ClusterList, error) {
b, err := getRawList(t.context, t)
func (t *clusterManager) getClusterList() (*cluster.ClusterList, error) {
b, err := getRawList(t.context.clientset, t.Name(), t.context.namespace)
if err != nil {
return nil, err
}
@@ -139,8 +161,13 @@ func (t *clusterTPR) getClusterList() (*cluster.ClusterList, error) {
return clusters, nil
}
func (t *clusterTPR) Watch() error {
logger.Infof("start watching %s tpr: %s", t.Name(), t.watchVersion)
func (t *clusterManager) EndWatch() {
logger.Infof("stop watching cluster tpr")
t.tracker.stop()
}
func (t *clusterManager) BeginWatch() error {
logger.Infof("start watching cluster tpr: %s", t.watchVersion)
defer t.tracker.stop()
eventCh, errCh := t.watch()
@@ -148,7 +175,7 @@ func (t *clusterTPR) Watch() error {
go func() {
timer := k8sutil.NewPanicTimer(
time.Minute,
fmt.Sprintf("unexpected long blocking (> 1 Minute) when handling %s event", t.Name()))
"unexpected long blocking (> 1 Minute) when handling cluster event")
for event := range eventCh {
timer.Start()
@@ -187,7 +214,7 @@ func (t *clusterTPR) Watch() error {
// 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 (t *clusterTPR) watch() (<-chan *clusterEvent, <-chan error) {
func (t *clusterManager) watch() (<-chan *clusterEvent, <-chan error) {
eventCh := make(chan *clusterEvent)
// On unexpected error case, the operator should exit
errCh := make(chan error, 1)
@@ -207,8 +234,8 @@ func (t *clusterTPR) watch() (<-chan *clusterEvent, <-chan error) {
return eventCh, errCh
}
func (t *clusterTPR) watchOuterTPR(eventCh chan *clusterEvent, errCh chan error) error {
resp, err := watchTPR(t.context, t.Name(), t.watchVersion)
func (t *clusterManager) watchOuterTPR(eventCh chan *clusterEvent, errCh chan error) error {
resp, err := watchTPR(t.context, t.Name(), t.context.namespace, t.watchVersion)
if err != nil {
errCh <- err
return err
@@ -238,7 +265,7 @@ func (t *clusterTPR) watchOuterTPR(eventCh chan *clusterEvent, errCh chan error)
}
}
func (t *clusterTPR) checkStaleCache() (bool, error) {
func (t *clusterManager) checkStaleCache() (bool, error) {
clusterList, err := t.getClusterList()
if err == nil && !t.isClustersCacheStale(clusterList.Items) {
t.watchVersion = clusterList.Metadata.ResourceVersion
@@ -248,7 +275,7 @@ func (t *clusterTPR) checkStaleCache() (bool, error) {
return true, err
}
func (t *clusterTPR) getRookClient(namespace string) (rookclient.RookRestClient, error) {
func (t *clusterManager) getRookClient(namespace string) (rookclient.RookRestClient, error) {
t.Lock()
defer t.Unlock()
if c, ok := t.clusters[namespace]; ok {
pkg/operator/cluster/pool.go
+13 -13
View File
@@ -33,8 +33,8 @@ const (
)
type Pool struct {
Metadata v1.ObjectMeta `json:"metadata,omitempty"`
PoolSpec `json:"spec"`
v1.ObjectMeta `json:"metadata,omitempty"`
PoolSpec `json:"spec"`
}
// Instantiate a new pool
@@ -46,26 +46,26 @@ func NewPool(spec PoolSpec) *Pool {
func (p *Pool) Create(rclient rookclient.RookRestClient) error {
// validate the pool settings
if err := p.validate(); err != nil {
return fmt.Errorf("invalid pool %s arguments. %+v", p.PoolSpec.Name, err)
return fmt.Errorf("invalid pool %s arguments. %+v", p.Name, err)
}
// check if the pool already exists
exists, err := p.exists(rclient)
if err == nil && exists {
logger.Infof("pool %s already exists in namespace %s", p.PoolSpec.Name, p.PoolSpec.Namespace)
logger.Infof("pool %s already exists in namespace %s", p.Name, p.Namespace)
return nil
}
// create the pool
pool := model.Pool{Name: p.PoolSpec.Name}
pool := model.Pool{Name: p.Name}
r := p.replication()
if r != nil {
logger.Infof("creating pool %s in namespace %s with replicas %d", p.PoolSpec.Name, p.Namespace, r.Count)
logger.Infof("creating pool %s in namespace %s with replicas %d", p.Name, p.Namespace, r.Count)
pool.ReplicationConfig.Size = r.Count
pool.Type = model.Replicated
} else {
ec := p.erasureCode()
logger.Infof("creating pool %s in namespace %s. coding chunks = %d, data chunks = %d", p.PoolSpec.Name, p.Namespace, ec.CodingChunks, ec.DataChunks)
logger.Infof("creating pool %s in namespace %s. coding chunks = %d, data chunks = %d", p.Name, p.Namespace, ec.CodingChunks, ec.DataChunks)
pool.ErasureCodedConfig.CodingChunkCount = ec.CodingChunks
pool.ErasureCodedConfig.DataChunkCount = ec.DataChunks
pool.Type = model.ErasureCoded
@@ -73,10 +73,10 @@ func (p *Pool) Create(rclient rookclient.RookRestClient) error {
info, err := rclient.CreatePool(pool)
if err != nil {
return fmt.Errorf("failed to create pool %s. %+v", p.PoolSpec.Name, err)
return fmt.Errorf("failed to create pool %s. %+v", p.Name, err)
}
logger.Infof("created pool %s. %s", p.PoolSpec.Name, info)
logger.Infof("created pool %s. %s", p.Name, info)
return nil
}
@@ -88,7 +88,7 @@ func (p *Pool) Delete(rclient rookclient.RookRestClient) error {
return nil
}
logger.Infof("TODO: delete pool %s from namespace %s", p.PoolSpec.Name, p.PoolSpec.Namespace)
logger.Infof("TODO: delete pool %s from namespace %s", p.Name, p.Namespace)
//return p.client.DeletePool(p.PoolSpec.Name)
return nil
}
@@ -100,7 +100,7 @@ func (p *Pool) exists(rclient rookclient.RookRestClient) (bool, error) {
return false, err
}
for _, pool := range pools {
if pool.Name == p.PoolSpec.Name {
if pool.Name == p.Name {
return true, nil
}
}
@@ -109,10 +109,10 @@ func (p *Pool) exists(rclient rookclient.RookRestClient) (bool, error) {
// Validate the pool arguments
func (p *Pool) validate() error {
if p.PoolSpec.Name == "" {
if p.Name == "" {
return fmt.Errorf("missing name")
}
if p.PoolSpec.Namespace == "" {
if p.Namespace == "" {
return fmt.Errorf("missing namespace")
}
if p.replication() != nil && p.erasureCode() != nil {
pkg/operator/cluster/pool_test.go
+27 -30
View File
@@ -21,6 +21,8 @@ package cluster
import (
"testing"
"k8s.io/client-go/pkg/api/v1"
"github.com/rook/rook/pkg/model"
"github.com/rook/rook/pkg/rook/test"
"github.com/stretchr/testify/assert"
@@ -28,47 +30,46 @@ import (
func TestValidatePool(t *testing.T) {
// must specify some replication or EC settings
spec := PoolSpec{Name: "mypool", Namespace: "myns"}
err := NewPool(spec).validate()
p := Pool{ObjectMeta: v1.ObjectMeta{Name: "mypool", Namespace: "myns"}}
err := p.validate()
assert.NotNil(t, err)
// must specify name
spec = PoolSpec{Namespace: "myns"}
err = NewPool(spec).validate()
p = Pool{ObjectMeta: v1.ObjectMeta{Namespace: "myns"}}
err = p.validate()
assert.NotNil(t, err)
// must specify namespace
spec = PoolSpec{Name: "mypool"}
err = NewPool(spec).validate()
p = Pool{ObjectMeta: v1.ObjectMeta{Name: "mypool"}}
err = p.validate()
assert.NotNil(t, err)
// must not specify both replication and EC settings
spec = PoolSpec{Name: "mypool", Namespace: "myns"}
spec.Replication.Count = 1
spec.ErasureCoding.CodingChunks = 2
spec.ErasureCoding.DataChunks = 3
err = NewPool(spec).validate()
p = Pool{ObjectMeta: v1.ObjectMeta{Name: "mypool", Namespace: "myns"}}
p.Replication.Count = 1
p.ErasureCoding.CodingChunks = 2
p.ErasureCoding.DataChunks = 3
err = p.validate()
assert.NotNil(t, err)
// succeed with replication settings
spec = PoolSpec{Name: "mypool", Namespace: "myns"}
spec.Replication.Count = 1
err = NewPool(spec).validate()
p = Pool{ObjectMeta: v1.ObjectMeta{Name: "mypool", Namespace: "myns"}}
p.Replication.Count = 1
err = p.validate()
assert.Nil(t, err)
// succeed with ec settings
spec = PoolSpec{Name: "mypool", Namespace: "myns"}
spec.ErasureCoding.CodingChunks = 1
spec.ErasureCoding.DataChunks = 2
err = NewPool(spec).validate()
p = Pool{ObjectMeta: v1.ObjectMeta{Name: "mypool", Namespace: "myns"}}
p.ErasureCoding.CodingChunks = 1
p.ErasureCoding.DataChunks = 2
err = p.validate()
assert.Nil(t, err)
}
func TestCreatePool(t *testing.T) {
rclient := &test.MockRookRestClient{}
spec := PoolSpec{Name: "mypool", Namespace: "myns"}
spec.Replication.Count = 1
p := NewPool(spec)
p := Pool{ObjectMeta: v1.ObjectMeta{Name: "mypool", Namespace: "myns"}}
p.Replication.Count = 1
exists, err := p.exists(rclient)
assert.False(t, exists)
@@ -76,15 +77,13 @@ func TestCreatePool(t *testing.T) {
assert.Nil(t, err)
// fail if both replication and EC are specified
spec.ErasureCoding.CodingChunks = 2
spec.ErasureCoding.DataChunks = 2
p.PoolSpec = spec
p.ErasureCoding.CodingChunks = 2
p.ErasureCoding.DataChunks = 2
err = p.Create(rclient)
assert.NotNil(t, err)
// succeed with EC
spec.Replication.Count = 0
p.PoolSpec = spec
p.Replication.Count = 0
err = p.Create(rclient)
assert.Nil(t, err)
}
@@ -100,8 +99,7 @@ func TestDeletePool(t *testing.T) {
}
// delete a pool that exists
spec := PoolSpec{Name: "mypool", Namespace: "myns"}
p := NewPool(spec)
p := Pool{ObjectMeta: v1.ObjectMeta{Name: "mypool", Namespace: "myns"}}
exists, err := p.exists(rclient)
assert.Nil(t, err)
assert.True(t, exists)
@@ -109,8 +107,7 @@ func TestDeletePool(t *testing.T) {
assert.Nil(t, err)
// succeed even if the pool doesn't exist
spec = PoolSpec{Name: "otherpool", Namespace: "myns"}
p = NewPool(spec)
p = Pool{ObjectMeta: v1.ObjectMeta{Name: "otherpool", Namespace: "myns"}}
exists, err = p.exists(rclient)
assert.Nil(t, err)
assert.False(t, exists)
pkg/operator/cluster/spec.go
-7
View File
@@ -40,13 +40,6 @@ type Spec struct {
}
type PoolSpec struct {
// The name of the pool. Defined in the spec since the object metadata name must be unique across all namespaces,
// while you could have the same pool name created in multiple instances of rook.
Name string `json:"name"`
// The namespace where the pool will be created (required). A Rook cluster must be running in this namespace.
Namespace string `json:"namespace"`
// The replication settings
Replication ReplicationSpec `json:"replication"`
pkg/operator/operator.go
+15 -25
View File
@@ -22,7 +22,6 @@ import (
"errors"
"fmt"
"net/http"
"sync"
"time"
"k8s.io/client-go/kubernetes"
@@ -54,8 +53,11 @@ type context struct {
}
type Operator struct {
context *context
tprs []TPR
context *context
tprSchemes []tprScheme
// The TPR that is global to the kubernetes cluster.
// The cluster TPR is global because you create multiple clusers in k8s
clusterMgr *clusterManager
}
func New(host, namespace string, factory client.ConnectionFactory, clientset kubernetes.Interface) *Operator {
@@ -67,10 +69,13 @@ func New(host, namespace string, factory client.ConnectionFactory, clientset kub
retryDelay: 3,
maxRetries: 30,
}
cluster := newClusterTPR(context)
poolInitiator := newPoolInitiator(context)
clusterMgr := newClusterManager(context, []inclusterInitiator{poolInitiator})
schemes := []tprScheme{clusterMgr, poolInitiator}
return &Operator{
context: context,
tprs: []TPR{cluster, newPoolTPR(context, cluster)},
context: context,
clusterMgr: clusterMgr,
tprSchemes: schemes,
}
}
@@ -85,20 +90,7 @@ func (o *Operator) Run() error {
}
// watch for changes to the rook clusters
var wg sync.WaitGroup
wg.Add(len(o.tprs))
for _, tpr := range o.tprs {
go func(t TPR) {
defer wg.Done()
if err := t.Watch(); err != nil {
logger.Errorf("failed to watch tpr %s. %+v", t.Name(), err)
}
}(tpr)
}
// wait for all of the TPRs to complete before exiting. If they complete it means there was an error.
wg.Wait()
return nil
return o.clusterMgr.BeginWatch()
}
func (o *Operator) initResources() error {
@@ -108,15 +100,13 @@ func (o *Operator) initResources() error {
}
o.context.kubeHttpCli = httpCli.Client
err = createTPRs(o.context, o.tprs)
err = createTPRs(o.context, o.tprSchemes)
if err != nil {
return fmt.Errorf("failed to create TPR. %+v", err)
}
for _, tpr := range o.tprs {
if err := tpr.Load(); err != nil {
return fmt.Errorf("failed to load tpr %s. %+v", tpr.Name(), err)
}
if err := o.clusterMgr.Load(); err != nil {
return fmt.Errorf("failed to load cluster. %+v", err)
}
return nil
pkg/operator/pool.go
+46 -39
View File
@@ -25,31 +25,45 @@ import (
"net/http"
"time"
kwatch "k8s.io/client-go/pkg/watch"
"github.com/rook/rook/pkg/operator/cluster"
"github.com/rook/rook/pkg/operator/k8sutil"
rookclient "github.com/rook/rook/pkg/rook/client"
kwatch "k8s.io/client-go/pkg/watch"
)
type poolTPR struct {
context *context
type poolInitiator struct {
context *context
}
type poolManager struct {
name string
namespace string
watchVersion string
cluster *clusterTPR
context *context
rclient rookclient.RookRestClient
}
func newPoolTPR(context *context, cluster *clusterTPR) *poolTPR {
return &poolTPR{context: context, cluster: cluster}
func newPoolInitiator(context *context) *poolInitiator {
return &poolInitiator{context: context}
}
func (t *poolTPR) Name() string {
func (p *poolInitiator) Create(clusterMgr *clusterManager, namespace string) (tprManager, error) {
rclient, err := clusterMgr.getRookClient(namespace)
if err != nil {
return nil, fmt.Errorf("failed to get api client for pool tpr in namespace %s. %+v", namespace, err)
}
return &poolManager{context: p.context, name: p.Name(), namespace: namespace, rclient: rclient}, nil
}
func (t *poolInitiator) Name() string {
return "rookpool"
}
func (t *poolTPR) Description() string {
func (t *poolInitiator) Description() string {
return "Managed Rook pools"
}
func (t *poolTPR) Load() error {
func (t *poolManager) Load() error {
// Check if pools have all been created
logger.Info("finding existing pools...")
poolList, err := t.getPoolList()
@@ -61,12 +75,8 @@ func (t *poolTPR) Load() error {
for i := range poolList.Items {
// ensure the pool exists
p := poolList.Items[i]
rclient, err := t.cluster.getRookClient(p.Namespace)
if err != nil {
return fmt.Errorf("failed to get rook client for namespace %s. %+v", p.Namespace, err)
}
if err := p.Create(rclient); err != nil {
logger.Warningf("failed to check that pool %s exists in namespace %s. %+v", p.PoolSpec.Name, p.PoolSpec.Namespace, err)
if err := p.Create(t.rclient); err != nil {
logger.Warningf("failed to check that pool %s exists in namespace %s. %+v", p.Name, p.Namespace, err)
}
}
@@ -74,8 +84,8 @@ func (t *poolTPR) Load() error {
return nil
}
func (t *poolTPR) getPoolList() (*cluster.PoolList, error) {
b, err := getRawList(t.context, t)
func (t *poolManager) getPoolList() (*cluster.PoolList, error) {
b, err := getRawList(t.context.clientset, t.name, t.namespace)
if err != nil {
return nil, err
}
@@ -87,47 +97,44 @@ func (t *poolTPR) getPoolList() (*cluster.PoolList, error) {
return pools, nil
}
func (t *poolTPR) Watch() error {
logger.Infof("start watching %s tpr: %s", t.Name(), t.watchVersion)
func (t *poolManager) EndWatch() error {
logger.Infof("TODO: stop watching pool tpr")
return nil
}
func (t *poolManager) BeginWatch() error {
logger.Infof("start watching pool tpr: %s", t.watchVersion)
eventCh, errCh := t.watch()
go func() {
timer := k8sutil.NewPanicTimer(
time.Minute,
fmt.Sprintf("unexpected long blocking (> 1 Minute) when handling %s event", t.Name()))
"unexpected long blocking (> 1 Minute) when handling pool event")
for event := range eventCh {
timer.Start()
pool := event.Object
rclient, err := t.cluster.getRookClient(pool.PoolSpec.Namespace)
if err != nil {
logger.Errorf("failed %v pool %s. %+v", event.Type, pool.Name, err)
break
}
switch event.Type {
case kwatch.Added:
p := cluster.NewPool(pool.PoolSpec)
if err := p.Create(rclient); err != nil {
logger.Errorf("failed to create pool %s. %+v", pool.PoolSpec.Name, err)
if err := pool.Create(t.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
p := cluster.NewPool(pool.PoolSpec)
if err := p.Create(rclient); err != nil {
logger.Errorf("failed to create (modify) pool %s. %+v", pool.PoolSpec.Name, err)
if err := pool.Create(t.rclient); err != nil {
logger.Errorf("failed to create (modify) pool %s. %+v", pool.Name, err)
break
}
case kwatch.Deleted:
oldPool := cluster.NewPool(pool.PoolSpec)
err := oldPool.Delete(rclient)
err := pool.Delete(t.rclient)
if err != nil {
logger.Errorf("failed to delete pool %s. %+v", pool.PoolSpec.Name, err)
logger.Errorf("failed to delete pool %s. %+v", pool.Name, err)
break
}
}
@@ -143,7 +150,7 @@ func (t *poolTPR) Watch() error {
// 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 (t *poolTPR) watch() (<-chan *poolEvent, <-chan error) {
func (t *poolManager) watch() (<-chan *poolEvent, <-chan error) {
eventCh := make(chan *poolEvent)
// On unexpected error case, the operator should exit
errCh := make(chan error, 1)
@@ -163,8 +170,8 @@ func (t *poolTPR) watch() (<-chan *poolEvent, <-chan error) {
return eventCh, errCh
}
func (t *poolTPR) watchOuterTPR(eventCh chan *poolEvent, errCh chan error) error {
resp, err := watchTPR(t.context, t.Name(), t.watchVersion)
func (t *poolManager) watchOuterTPR(eventCh chan *poolEvent, errCh chan error) error {
resp, err := watchTPR(t.context, t.name, t.namespace, t.watchVersion)
if err != nil {
errCh <- err
return err
@@ -187,7 +194,7 @@ func (t *poolTPR) watchOuterTPR(eventCh chan *poolEvent, errCh chan error) error
}
logger.Debugf("rook pool event: %+v", ev)
t.watchVersion = ev.Object.Metadata.ResourceVersion
t.watchVersion = ev.Object.ResourceVersion
eventCh <- ev
}
}
pkg/operator/tpr.go
+24 -15
View File
@@ -26,6 +26,7 @@ import (
"github.com/rook/rook/pkg/operator/k8sutil"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/pkg/api/errors"
"k8s.io/client-go/pkg/api/unversioned"
"k8s.io/client-go/pkg/api/v1"
@@ -37,18 +38,26 @@ const (
tprVersion = "v1alpha1"
)
type TPR interface {
type tprScheme interface {
Name() string
Description() string
Load() error
Watch() error
}
func qualifiedName(tpr TPR) string {
type inclusterInitiator interface {
Create(clusterMgr *clusterManager, namespace string) (tprManager, error)
}
type tprManager interface {
Load() error
BeginWatch() error
EndWatch() error
}
func qualifiedName(tpr tprScheme) string {
return fmt.Sprintf("%s.%s", tpr.Name(), tprGroup)
}
func createTPRs(context *context, tprs []TPR) error {
func createTPRs(context *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)
@@ -64,7 +73,7 @@ func createTPRs(context *context, tprs []TPR) error {
return nil
}
func createTPR(context *context, tpr TPR) error {
func createTPR(context *context, tpr tprScheme) error {
logger.Infof("creating %s TPR", tpr.Name())
r := &v1beta1.ThirdPartyResource{
ObjectMeta: v1.ObjectMeta{
@@ -85,9 +94,9 @@ func createTPR(context *context, tpr TPR) error {
return nil
}
func waitForTPRInit(context *context, tpr TPR) error {
func waitForTPRInit(context *context, tpr tprScheme) error {
restcli := context.clientset.CoreV1().RESTClient()
uri := tprURI(context, tpr.Name())
uri := tprURI(tpr.Name(), context.namespace)
return k8sutil.Retry(time.Duration(context.retryDelay)*time.Second, context.maxRetries, func() (bool, error) {
_, err := restcli.Get().RequestURI(uri).DoRaw()
if err != nil {
@@ -101,18 +110,18 @@ func waitForTPRInit(context *context, tpr TPR) error {
})
}
func watchTPR(context *context, name string, resourceVersion string) (*http.Response, error) {
func watchTPR(context *context, name, namespace, resourceVersion string) (*http.Response, error) {
return context.kubeHttpCli.Get(fmt.Sprintf("%s/%s?watch=true&resourceVersion=%s",
context.masterHost, tprURI(context, name), resourceVersion))
context.masterHost, tprURI(name, namespace), resourceVersion))
}
func tprURI(context *context, name string) string {
return fmt.Sprintf("/apis/%s/%s/namespaces/%s/%ss", tprGroup, tprVersion, context.namespace, name)
func tprURI(name, namespace string) string {
return fmt.Sprintf("/apis/%s/%s/namespaces/%s/%ss", tprGroup, tprVersion, namespace, name)
}
func getRawList(context *context, tpr TPR) ([]byte, error) {
restcli := context.clientset.CoreV1().RESTClient()
return restcli.Get().RequestURI(tprURI(context, tpr.Name())).DoRaw()
func getRawList(clientset kubernetes.Interface, name, namespace string) ([]byte, error) {
restcli := clientset.CoreV1().RESTClient()
return restcli.Get().RequestURI(tprURI(name, namespace)).DoRaw()
}
func handlePollEventResult(status *unversioned.Status, errIn error, checkStaleCache func() (bool, error), errCh chan error) (done bool, err error) {