diff --git a/demo/kubernetes/rook-pool.yaml b/demo/kubernetes/rook-pool.yaml
index b86cb614d..5cea07cc8 100644
--- a/demo/kubernetes/rook-pool.yaml
+++ b/demo/kubernetes/rook-pool.yaml
@@ -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.
diff --git a/pkg/operator/cluster.go b/pkg/operator/cluster.go
index b77f6d5e7..cda1bd53a 100644
--- a/pkg/operator/cluster.go
+++ b/pkg/operator/cluster.go
@@ -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 {
diff --git a/pkg/operator/cluster/pool.go b/pkg/operator/cluster/pool.go
index 17fe6df3a..80ea5e461 100644
--- a/pkg/operator/cluster/pool.go
+++ b/pkg/operator/cluster/pool.go
@@ -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 {
diff --git a/pkg/operator/cluster/pool_test.go b/pkg/operator/cluster/pool_test.go
index c5d543f84..945f1b585 100644
--- a/pkg/operator/cluster/pool_test.go
+++ b/pkg/operator/cluster/pool_test.go
@@ -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)
diff --git a/pkg/operator/cluster/spec.go b/pkg/operator/cluster/spec.go
index 21eeefb82..842e4ed05 100644
--- a/pkg/operator/cluster/spec.go
+++ b/pkg/operator/cluster/spec.go
@@ -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"`
 
diff --git a/pkg/operator/operator.go b/pkg/operator/operator.go
index 9ed301f89..4401af134 100644
--- a/pkg/operator/operator.go
+++ b/pkg/operator/operator.go
@@ -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
diff --git a/pkg/operator/pool.go b/pkg/operator/pool.go
index eb7f68451..8d3b3cded 100644
--- a/pkg/operator/pool.go
+++ b/pkg/operator/pool.go
@@ -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
 	}
 }
diff --git a/pkg/operator/tpr.go b/pkg/operator/tpr.go
index 3176aefc6..0c41f8ac6 100644
--- a/pkg/operator/tpr.go
+++ b/pkg/operator/tpr.go
@@ -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) {