Repository: kubernetes-sigs/kubefed
Pull request #1292 (2020-10-02 20:24:23)

feat: collect remote resource status when enabled
<!--  Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, read our contributor guidelines https://git.k8s.io/community/contributors/guide#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. If you want *faster* PR reviews, read how: https://git.k8s.io/community/contributors/guide/pull-requests.md#best-practices-for-faster-reviews
3. Add an entry to CHANGELOG.md if the PR represents a user-visible change.
4. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

**What this PR does / why we need it**:

This PR aims to implement this feature https://github.com/kubernetes-sigs/kubefed/blob/ba8f9a44e85951e460454ab7875cb0a551750124/docs/keps/20200619-federated-resource-status.md. This work is a first iteration towards improving the troubleshooting of the status of federated resources.

In this branch, we add a new property to the FederatedResources `remoteStatus` which stores the status of the created object on each federated cluster. An example of a FederatedDeployment and its status is shown below:

```
- apiVersion: types.kubefed.io/v1beta1
  kind: FederatedDeployment
  metadata:
    finalizers:
    - kubefed.io/sync-controller
    generation: 1
    name: asystem
    namespace: asystem
    resourceVersion: "70174497"
  spec:
    placement:
      clusters:
      - name: cluster1
    template:
      metadata:
        labels:
          app: nginx
      spec:
        replicas: 3
        selector:
          matchLabels:
            app: nginx
        template:
          metadata:
            labels:
              app: nginx
          spec:
            containers:
            - image: nginx
              name: nginx
  status:
    clusters:
    - name: "cluster1"
      remoteStatus:
        availableReplicas: 3
        conditions:
        - lastTransitionTime: "2020-09-18T11:07:55Z"
          lastUpdateTime: "2020-09-18T11:18:31Z"
          message: ReplicaSet "asystem-f89759699" has successfully progressed.
          reason: NewReplicaSetAvailable
          status: "True"
          type: Progressing
        - lastTransitionTime: "2020-09-24T05:42:11Z"
          lastUpdateTime: "2020-09-24T05:42:11Z"
          message: Deployment has minimum availability.
          reason: MinimumReplicasAvailable
          status: "True"
          type: Available
        observedGeneration: 3
        readyReplicas: 3
        replicas: 3
        updatedReplicas: 3
    conditions:
    - lastTransitionTime: "2020-05-25T20:23:59Z"
      lastUpdateTime: "2020-05-25T20:23:59Z"
      status: "True"
      type: "Propagation"
```

To enable this feature, you need to add the feature gate in the kubefedconfig spec:

```
  featureGates:
  - name: RawResourceStatusCollection
    configuration: "Enabled"
```

In addition to that step, you will need to enable the `statusCollection` for the FederatedTypeConfig resources you wish to collect the `remoteStatus`. To do so, you set the property `statusCollection` to `Enabled`.

```
spec:
  federatedType:
    ...
  propagation: Enabled
  statusCollection: Enabled
  targetType:
    ...
```

**Which issue(s) this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close the issue(s) when PR gets merged)*:
Fixes #

**Special notes for your reviewer**:
I am testing some changes in the test suite so additional polishing actions are expected. 

Obviously this feature could potentially cause some performance degradation if our scenario federates many resources and we enable the status collection for all them. We are working on measuring and improving this behavior.


k8s-ci-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/kubefed/pull/1292#" title="Author self-approved">hectorj2f</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fkubefed).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/kubernetes-sigs/kubefed/blob/master/OWNERS)~~ [hectorj2f]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
hectorj2f: I'll have a look at the tests.
hectorj2f: It seems the in-memory controllers test is failing 🔍 .
jimmidyson: @hectorj2f I noticed that the default federated CRDs don't have `RemoteStatus` in their CRD, plus this is also not present when you `kubefedctl enable` a new resource for federation. This will also affect the migration to `apiextensions.k8s.io/v1` (cc @makkes for #1296).
k8s-ci-robot: @hectorj2f: PR needs rebase.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
k8s-ci-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/kubefed/pull/1292#" title="Author self-approved">hectorj2f</a>*, *<a href="https://github.com/kubernetes-sigs/kubefed/pull/1292#pullrequestreview-503801334" title="Approved">makkes</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fkubefed).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/kubernetes-sigs/kubefed/blob/master/OWNERS)~~ [hectorj2f]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
irfanurrehman: @hectorj2f I see that you have pushed new changes. Have you completed your review rework?
hectorj2f: @irfanurrehman Yes. But I want to share some graphs and numbers from a test we performed to compare existing vs current approach. I'll ping you this week, thanks for keeping one eye on the PR.

Patch:
diff --git a/charts/kubefed/README.md b/charts/kubefed/README.md
index ada0a5e6..c4b4b730 100644
--- a/charts/kubefed/README.md
+++ b/charts/kubefed/README.md
@@ -103,10 +103,10 @@ chart and their default values.
 | controllermanager.image               | Name of the KubeFed image.                                                                                                                                                                  | kubefed                         |
 | controllermanager.tag                 | Tag of the KubeFed image.                                                                                                                                                                   | latest                          |
 | controllermanager.imagePullPolicy     | Image pull policy.                                                                                                                                                                          | IfNotPresent                    |
-| controllermanager.imagePullSecrets    | Image pull secrets.                                                                                                                                                                         | [] 
 | controllermanager.commonTolerations   | Tolerations for all the pods.                                                                                                                                                               | []                              |
 | controllermanager.commonNodeSelector     | Node selector for all the pods.                                                                                                                                                          | {}                              |
 | controllermanager.featureGates.PushReconciler               | Push reconciler feature.                                                                                                                                              | true                            |
+| controllermanager.featureGates.RawResourceStatusCollection               | Raw collection of resource status on target clusters feature.                                                                                                                                              | false                            |
 | controllermanager.featureGates.SchedulerPreferences         | Scheduler preferences feature.                                                                                                                                        | true                            |
 | controllermanager.featureGates.CrossClusterServiceDiscovery | Cross cluster service discovery feature.                                                                                                                              | false                           |
 | controllermanager.featureGates.FederatedIngress             | Federated ingress feature.                                                                                                                                            | false                           |
diff --git a/charts/kubefed/charts/controllermanager/templates/deployments.yaml b/charts/kubefed/charts/controllermanager/templates/deployments.yaml
index 8d92584b..5a3e98b9 100644
--- a/charts/kubefed/charts/controllermanager/templates/deployments.yaml
+++ b/charts/kubefed/charts/controllermanager/templates/deployments.yaml
@@ -24,10 +24,6 @@ spec:
         rollme: {{ randAlphaNum 5 | quote }}
       {{- end }}
     spec:
-      {{- if .Values.imagePullSecrets }}
-      imagePullSecrets:
-{{ toYaml .Values.imagePullSecrets | indent 8 }}
-      {{- end }}
       nodeSelector:
         {{- toYaml .Values.commonNodeSelector | nindent 8 }}
       tolerations:
@@ -90,10 +86,6 @@ spec:
         rollme: {{ randAlphaNum 5 | quote }}
       {{- end }}
     spec:
-      {{- if .Values.imagePullSecrets }}
-      imagePullSecrets:
-{{ toYaml .Values.imagePullSecrets | indent 8 }}
-      {{- end }}
       nodeSelector:
         {{- toYaml .Values.commonNodeSelector | nindent 8 }}
       tolerations:
diff --git a/charts/kubefed/charts/controllermanager/templates/kubefedconfig.yaml b/charts/kubefed/charts/controllermanager/templates/kubefedconfig.yaml
index f333f592..293175cf 100644
--- a/charts/kubefed/charts/controllermanager/templates/kubefedconfig.yaml
+++ b/charts/kubefed/charts/controllermanager/templates/kubefedconfig.yaml
@@ -24,6 +24,8 @@ spec:
 {{- if .Values.featureGates }}
   - name: PushReconciler
     configuration: {{ .Values.featureGates.PushReconciler | default "Enabled" | quote }}
+  - name: RawResourceStatusCollection
+    configuration: {{ .Values.featureGates.RawResourceStatusCollection | default "Disabled" | quote }}
   - name: SchedulerPreferences
     configuration: {{ .Values.featureGates.SchedulerPreferences | default "Enabled" | quote }}
   - name: CrossClusterServiceDiscovery
diff --git a/charts/kubefed/crds/crds.yaml b/charts/kubefed/crds/crds.yaml
index 1ceff9bf..d347cbe9 100644
--- a/charts/kubefed/crds/crds.yaml
+++ b/charts/kubefed/crds/crds.yaml
@@ -89,6 +89,9 @@ spec:
                   properties:
                     name:
                       type: string
+                    remoteStatus:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
                     status:
                       type: string
                   required:
@@ -219,6 +222,9 @@ spec:
                   properties:
                     name:
                       type: string
+                    remoteStatus:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
                     status:
                       type: string
                   required:
@@ -351,6 +357,9 @@ spec:
                   properties:
                     name:
                       type: string
+                    remoteStatus:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
                     status:
                       type: string
                   required:
@@ -481,6 +490,9 @@ spec:
                   properties:
                     name:
                       type: string
+                    remoteStatus:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
                     status:
                       type: string
                   required:
@@ -609,6 +621,9 @@ spec:
                   properties:
                     name:
                       type: string
+                    remoteStatus:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
                     status:
                       type: string
                   required:
@@ -739,6 +754,9 @@ spec:
                   properties:
                     name:
                       type: string
+                    remoteStatus:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
                     status:
                       type: string
                   required:
@@ -871,6 +889,9 @@ spec:
                   properties:
                     name:
                       type: string
+                    remoteStatus:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
                     status:
                       type: string
                   required:
@@ -999,6 +1020,9 @@ spec:
                   properties:
                     name:
                       type: string
+                    remoteStatus:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
                     status:
                       type: string
                   required:
@@ -1129,6 +1153,9 @@ spec:
                   properties:
                     name:
                       type: string
+                    remoteStatus:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
                     status:
                       type: string
                   required:
@@ -1259,6 +1286,9 @@ spec:
                   properties:
                     name:
                       type: string
+                    remoteStatus:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
                     status:
                       type: string
                   required:
diff --git a/charts/kubefed/values.yaml b/charts/kubefed/values.yaml
index 97636a7f..f229319d 100644
--- a/charts/kubefed/values.yaml
+++ b/charts/kubefed/values.yaml
@@ -25,6 +25,7 @@ controllermanager:
     SchedulerPreferences:
     CrossClusterServiceDiscovery:
     FederatedIngress:
+    RawResourceStatusCollection:
 
   ## common node selector
   commonNodeSelector: {}
@@ -38,10 +39,6 @@ controllermanager:
   #   value: "value"
   #   effect: "NoSchedule"
 
-  ## Ref: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
-  imagePullSecrets: []
-  # - name: secretName
-
   controller:
     annotations: {}
     replicaCount: 2
diff --git a/cmd/controller-manager/app/controller-manager.go b/cmd/controller-manager/app/controller-manager.go
index 080954af..2ee8ed58 100644
--- a/cmd/controller-manager/app/controller-manager.go
+++ b/cmd/controller-manager/app/controller-manager.go
@@ -213,6 +213,11 @@ func startControllers(opts *options.Options, stopChan <-chan struct{}) {
 	}
 
 	if utilfeature.DefaultFeatureGate.Enabled(features.PushReconciler) {
+		if utilfeature.DefaultFeatureGate.Enabled(features.RawResourceStatusCollection) {
+			opts.Config.RawResourceStatusCollection = true
+			klog.Info("Enabling RawResourceStatusCollection for all the enabled federated resources")
+		}
+
 		if err := federatedtypeconfig.StartController(opts.Config, stopChan); err != nil {
 			klog.Fatalf("Error starting federated type config controller: %v", err)
 		}
diff --git a/config/kubefedconfig.yaml b/config/kubefedconfig.yaml
index c9dbc0ea..3e0142f5 100644
--- a/config/kubefedconfig.yaml
+++ b/config/kubefedconfig.yaml
@@ -16,6 +16,8 @@ spec:
   featureGates:
   - name: PushReconciler
     configuration: "Enabled"
+  - name: RawResourceStatusCollection
+    configuration: "Enabled"
   - name: SchedulerPreferences
     configuration: "Enabled"
   - name: CrossClusterServiceDiscovery
diff --git a/docs/development.md b/docs/development.md
index 514bf8e7..6d3a47b1 100644
--- a/docs/development.md
+++ b/docs/development.md
@@ -98,6 +98,8 @@ types enabled by default:
 1. a label update is reflected in the objects stored in the target
    clusters.
 1. a placement update for the object is reflected in the target clusters.
+1. optionally if `RawResourceStatusCollection` feature is enabled, tests check the value
+of the `remoteStatus` field of federated resources.
 1. deleted resources are removed from the target clusters.
 
 The read operation is implicit.
diff --git a/docs/environments/kind.md b/docs/environments/kind.md
index 13fe3bc5..3e926ff2 100644
--- a/docs/environments/kind.md
+++ b/docs/environments/kind.md
@@ -43,19 +43,6 @@ script if you'd like to change the default:
 NUM_CLUSTERS=<num> ./scripts/create-clusters.sh
 ```
 
-The `KIND_TAG` is `v1.19.4@sha256:796d09e217d93bed01ecf8502633e48fd806fe42f9d02fdd468b81cd4e3bd40b` by default.
-Image `kindest/node:v1.19.4@sha256:796d09e217d93bed01ecf8502633e48fd806fe42f9d02fdd468b81cd4e3bd40b` is used as
-node docker image for booting the cluster.
-
-You can use `KIND_IMAGE` or `KIND_TAG` to specify the image as you want.
-```bash
-KIND_TAG=v1.18.8 ./scripts/create-clusters.sh
-```
-
-```bash
-KIND_IMAGE=kindest/node:v1.18.8 ./scripts/create-clusters.sh
-```
-
 ## Delete Clusters
 
 Run the following command to delete `2` `kind` clusters:
diff --git a/docs/keps/20200619-federated-resource-status.md b/docs/keps/20200619-federated-resource-status.md
index 0beaec76..f8108c01 100644
--- a/docs/keps/20200619-federated-resource-status.md
+++ b/docs/keps/20200619-federated-resource-status.md
@@ -115,9 +115,9 @@ type GenericFederatedResource struct {
 }
 
 type GenericClusterStatus struct {
-	Name   string            `json:"name"`
-	Status PropagationStatus `json:"status,omitempty"`
-
+  Name   string                  `json:"name"`
+  Status PropagationStatus       `json:"status,omitempty"`
+  RemoteStatus interface{}       `json:"remoteStatus,omitempty"`
   Conditions []*metav1.Condition `json:"conditions,omitempty"`
 }
 
@@ -365,11 +365,6 @@ to expect a common schema for `.status.conditions` and share golang logic for co
 By following this approach, kubefed would be able to properly consume and report the status
 of any federated resource by checking the `status.conditions` (e.g `Ready=True`) fields.
 
-
-**Note that**, this proposed solution might include additional flags to the kubefed control-plane
-components to avoid blowing out the control-plane due to frequent and concurrent API calls to
-update the status of the federated resources.
-
 Obviously users might want to be able to enable/disable the definition of:
 
 * Which condition to look for each federated resource to determine its readiness, e.g. `Ready=True`, `Deployed=True`.
@@ -379,8 +374,99 @@ This might be especially useful for custom resource types without a `IsReady` st
 By raw status, the system understand to show the status of all the federated resources `Ready` and not `Ready`.
 This can have an impact in the performance, so this should be configured with precautions.
 
-To do so, the system exposes properties as part of each `FederatedTypeConfig` to define
-the desired behavior at federated resource type.
+**Note that**, this proposed solution includes additional flags to the kubefed control-plane
+components to limit the calls per second to the API, which could cause performance degradations
+to the control-plane when updating the status of the federated resources.
+
+In addition to the `conditions` field, this new feature adds a new property `remoteStatus` that holds
+the status of the created resource created in the kubefed cluster.The `remoteStatus` is an unstructured
+object due to the unknown structure of the status of the target resources.
+To enable the collection of the status of federated resources, a new feature gate, disabled by default, is added to the current list of features.
+Likewise the user has to explicitly enable the collection at `FederatedTypeConfig` level. To do so, the `FederatedTypeConfig.statusCollection` field
+has to be set to `Enabled`.
+
+As an example, the following property needs to be enabled in the `FederatedDeployment`:
+
+```yaml
+spec:
+  federatedType:
+    group: types.kubefed.io
+    kind: FederatedDeployment
+    pluralName: federateddeployments
+    scope: Namespaced
+    version: v1beta1
+  propagation: Enabled
+  statusCollection: Enabled
+  targetType:
+    group: apps
+    kind: Deployment
+    pluralName: deployments
+    scope: Namespaced
+    version: v1
+```
+
+An example of a `FederatedDeployment` federated in a `cluster1` where the feature
+collected the resource remote status `remoteStatus`.
+
+```yaml
+- apiVersion: types.kubefed.io/v1beta1
+  kind: FederatedDeployment
+  metadata:
+    finalizers:
+    - kubefed.io/sync-controller
+    generation: 1
+    name: asystem
+    namespace: asystem
+    resourceVersion: "70174497"
+  spec:
+    placement:
+      clusters:
+      - name: cluster1
+    template:
+      metadata:
+        labels:
+          app: nginx
+      spec:
+        replicas: 3
+        selector:
+          matchLabels:
+            app: nginx
+        template:
+          metadata:
+            labels:
+              app: nginx
+          spec:
+            containers:
+            - image: nginx
+              name: nginx
+  status:
+    clusters:
+    - name: "cluster1"
+      remoteStatus:
+        availableReplicas: 3
+        conditions:
+        - lastTransitionTime: "2020-09-18T11:07:55Z"
+          lastUpdateTime: "2020-09-18T11:18:31Z"
+          message: ReplicaSet "asystem-f89759699" has successfully progressed.
+          reason: NewReplicaSetAvailable
+          status: "True"
+          type: Progressing
+        - lastTransitionTime: "2020-09-24T05:42:11Z"
+          lastUpdateTime: "2020-09-24T05:42:11Z"
+          message: Deployment has minimum availability.
+          reason: MinimumReplicasAvailable
+          status: "True"
+          type: Available
+        observedGeneration: 3
+        readyReplicas: 3
+        replicas: 3
+        updatedReplicas: 3
+    conditions:
+    - lastTransitionTime: "2020-05-25T20:23:59Z"
+      lastUpdateTime: "2020-05-25T20:23:59Z"
+      status: "True"
+      type: "Propagation"
+```
 
 ### User Stories
 
diff --git a/example/config/kubefedconfig.yaml b/example/config/kubefedconfig.yaml
index 2a0f56c5..f3efbc4c 100644
--- a/example/config/kubefedconfig.yaml
+++ b/example/config/kubefedconfig.yaml
@@ -17,6 +17,8 @@ spec:
   featureGates:
   - name: PushReconciler
     configuration: "Enabled"
+  - name: RawResourceStatusCollection
+    configuration: "Enabled"
   - name: SchedulerPreferences
     configuration: "Enabled"
   - name: CrossClusterServiceDiscovery
diff --git a/go.mod b/go.mod
index c16d658d..0a3b464d 100644
--- a/go.mod
+++ b/go.mod
@@ -5,7 +5,7 @@ go 1.14
 require (
 	github.com/evanphx/json-patch v4.9.0+incompatible
 	github.com/ghodss/yaml v1.0.0
-	github.com/go-logr/logr v0.2.1 // indirect
+	github.com/go-logr/logr v0.3.0 // indirect
 	github.com/go-logr/zapr v0.2.0 // indirect
 	github.com/json-iterator/go v1.1.10
 	github.com/onsi/ginkgo v1.14.2
@@ -16,6 +16,8 @@ require (
 	github.com/spf13/cobra v1.1.1
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.6.1
+	go.uber.org/zap v1.16.0 // indirect
+	golang.org/x/text v0.3.4 // indirect
 	k8s.io/api v0.19.3
 	k8s.io/apiextensions-apiserver v0.19.3
 	k8s.io/apimachinery v0.19.3
diff --git a/go.sum b/go.sum
index 8d786f24..da24b91f 100644
--- a/go.sum
+++ b/go.sum
@@ -177,6 +177,8 @@ github.com/go-logr/logr v0.2.0 h1:QvGt2nLcHH0WK9orKa+ppBPAxREcH364nPUedEpK0TY=
 github.com/go-logr/logr v0.2.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=
 github.com/go-logr/logr v0.2.1 h1:fV3MLmabKIZ383XifUjFSwcoGee0v9qgPp8wy5svibE=
 github.com/go-logr/logr v0.2.1/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=
+github.com/go-logr/logr v0.3.0 h1:q4c+kbcR0d5rSurhBR8dIgieOaYpXtsdTYfx22Cu6rs=
+github.com/go-logr/logr v0.3.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=
 github.com/go-logr/zapr v0.1.0 h1:h+WVe9j6HAA01niTJPA/kKH0i7e0rLZBCwauQFcRE54=
 github.com/go-logr/zapr v0.1.0/go.mod h1:tabnROwaDl0UNxkVeFRbY8bwB37GwRv0P8lg6aAiEnk=
 github.com/go-logr/zapr v0.2.0 h1:v6Ji8yBW77pva6NkJKQdHLAJKrIJKRHz0RXwPqCHSR4=
@@ -606,10 +608,14 @@ go.uber.org/atomic v1.4.0 h1:cxzIVoETapQEqDhQu3QfnvXAV4AlzcvUCxkVUFw3+EU=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.5.0 h1:OI5t8sDa1Or+q8AeE+yKeB/SDYioSHAgcVljj9JIETY=
 go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
+go.uber.org/atomic v1.6.0 h1:Ezj3JGmsOnG1MoRWQkPBsKLe9DwWD9QeXzTRzzldNVk=
+go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
 go.uber.org/multierr v1.1.0 h1:HoEmRHQPVSqub6w2z2d2EOVs2fjyFRGyofhKuyDq0QI=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
 go.uber.org/multierr v1.3.0 h1:sFPn2GLc3poCkfrpIXGhBD2X0CMIo4Q/zSULXrj/+uc=
 go.uber.org/multierr v1.3.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+4=
+go.uber.org/multierr v1.5.0 h1:KCa4XfM8CWFCpxXRGok+Q0SS/0XBhMDbHHGABQLvD2A=
+go.uber.org/multierr v1.5.0/go.mod h1:FeouvMocqHpRaaGuG9EjoKcStLC43Zu/fmqdUMPcKYU=
 go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee h1:0mgffUl7nfd+FpvXMVz4IDEaUSmT1ysygQC7qYo7sG4=
 go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee/go.mod h1:vJERXedbb3MVM5f9Ejo0C68/HhF8uaILCdgjnY+goOA=
 go.uber.org/zap v1.8.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
@@ -617,6 +623,8 @@ go.uber.org/zap v1.10.0 h1:ORx85nbTijNz8ljznvCMR1ZBIPKFn3jQrag10X2AsuM=
 go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 go.uber.org/zap v1.13.0 h1:nR6NoDBgAf67s68NhaXbsojM+2gxp3S1hWkHDl27pVU=
 go.uber.org/zap v1.13.0/go.mod h1:zwrFLgMcdUuIBviXEYEH1YKNaOBnKXsx2IPda5bBwHM=
+go.uber.org/zap v1.16.0 h1:uFRZXykJGK9lLY4HtgSw44DnIcAM+kRBP7x5m+NpAOM=
+go.uber.org/zap v1.16.0/go.mod h1:MA8QOfq0BHJwdXa996Y4dYkAqRKB8/1K1QMMZVaNZjQ=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
@@ -755,6 +763,8 @@ golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.4 h1:0YWbFKbhXG/wIiuHDSKpS0Iy7FSA+u45VtBMfQcFTTc=
+golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 h1:SvFZT6jyqRaOeXpc5h/JSfZenJ2O330aBsf7JfSUXmQ=
diff --git a/pkg/apis/core/v1beta1/federatedtypeconfig_types.go b/pkg/apis/core/v1beta1/federatedtypeconfig_types.go
index cab26449..aac0aae9 100644
--- a/pkg/apis/core/v1beta1/federatedtypeconfig_types.go
+++ b/pkg/apis/core/v1beta1/federatedtypeconfig_types.go
@@ -207,18 +207,20 @@ func (f *FederatedTypeConfig) GetFederatedType() metav1.APIResource {
 	return apiResourceToMeta(f.Spec.FederatedType, f.GetFederatedNamespaced())
 }
 
+// TODO (hectorj2f): It should get deprecated once we move to the new status approach
+// because the type is the same as the target type.
 func (f *FederatedTypeConfig) GetStatusType() *metav1.APIResource {
 	if f.Spec.StatusType == nil {
 		return nil
 	}
+	// Return the original target type
 	metaAPIResource := apiResourceToMeta(*f.Spec.StatusType, f.Spec.StatusType.Namespaced())
 	return &metaAPIResource
 }
 
 func (f *FederatedTypeConfig) GetStatusEnabled() bool {
 	return f.Spec.StatusCollection != nil &&
-		*f.Spec.StatusCollection == StatusCollectionEnabled &&
-		f.Name == "services"
+		*f.Spec.StatusCollection == StatusCollectionEnabled
 }
 
 // TODO(font): This method should be removed from the interface i.e. remove
diff --git a/pkg/apis/core/v1beta1/validation/validation.go b/pkg/apis/core/v1beta1/validation/validation.go
index 270177df..dce9d364 100644
--- a/pkg/apis/core/v1beta1/validation/validation.go
+++ b/pkg/apis/core/v1beta1/validation/validation.go
@@ -310,7 +310,7 @@ func ValidateKubeFedConfig(kubeFedConfig, oldKubeFedConfig *v1beta1.KubeFedConfi
 			existingNames[gate.Name] = true
 
 			allErrs = append(allErrs, validateEnumStrings(gatesPath.Child("name"), string(gate.Name),
-				[]string{string(features.PushReconciler), string(features.SchedulerPreferences),
+				[]string{string(features.PushReconciler), string(features.RawResourceStatusCollection), string(features.SchedulerPreferences),
 					string(features.CrossClusterServiceDiscovery), string(features.FederatedIngress)})...)
 
 			allErrs = append(allErrs, validateEnumStrings(gatesPath.Child("configuration"), string(gate.Configuration),
diff --git a/pkg/controller/federatedtypeconfig/controller.go b/pkg/controller/federatedtypeconfig/controller.go
index 37e39aa9..990aa649 100644
--- a/pkg/controller/federatedtypeconfig/controller.go
+++ b/pkg/controller/federatedtypeconfig/controller.go
@@ -148,7 +148,10 @@ func (c *Controller) reconcile(qualifiedName util.QualifiedName) util.Reconcilia
 	corev1b1.SetFederatedTypeConfigDefaults(typeConfig)
 
 	syncEnabled := typeConfig.GetPropagationEnabled()
-	statusEnabled := typeConfig.GetStatusEnabled()
+	// NOTE (hectorj2f): RawResourceStatusCollection is a new feature and is
+	// Disabled by default. When RawResourceStatusCollection is enabled,
+	// the old mechanism to collect the service status of FederatedServices would be disabled.
+	statusControllerEnabled := !c.controllerConfig.RawResourceStatusCollection && c.isEnabledFederatedServiceStatusCollection(typeConfig)
 
 	limitedScope := c.controllerConfig.TargetNamespace != metav1.NamespaceAll
 	if limitedScope && syncEnabled && !typeConfig.GetNamespaced() {
@@ -225,8 +228,8 @@ func (c *Controller) reconcile(qualifiedName util.QualifiedName) util.Reconcilia
 		c.stopController(typeConfig.Name, syncStopChan)
 	}
 
-	startNewStatusController := !statusRunning && statusEnabled
-	stopStatusController := statusRunning && !statusEnabled
+	startNewStatusController := !statusRunning && statusControllerEnabled
+	stopStatusController := statusRunning && !statusControllerEnabled
 	if startNewStatusController {
 		if err := c.startStatusController(statusKey, typeConfig); err != nil {
 			runtime.HandleError(err)
@@ -433,3 +436,17 @@ func (c *Controller) reconcileOnNamespaceFTCUpdate() {
 		}
 	}
 }
+
+func (c *Controller) isEnabledFederatedServiceStatusCollection(tc *corev1b1.FederatedTypeConfig) bool {
+	if tc.GetStatusEnabled() && tc.Name == "services" {
+		federatedAPIResource := tc.GetFederatedType()
+		statusAPIResource := tc.GetStatusType()
+		if statusAPIResource == nil {
+			klog.Infof("Skipping status collection, status API resource is not defined for %q", federatedAPIResource.Kind)
+			return false
+		}
+		klog.Info("Federated Service Status collection is enabled")
+		return true
+	}
+	return false
+}
diff --git a/pkg/controller/sync/controller.go b/pkg/controller/sync/controller.go
index 0643eb02..ffa529ed 100644
--- a/pkg/controller/sync/controller.go
+++ b/pkg/controller/sync/controller.go
@@ -87,6 +87,8 @@ type KubeFedSyncController struct {
 	skipAdoptingResources bool
 
 	limitedScope bool
+
+	rawResourceStatusCollection bool
 }
 
 // StartKubeFedSyncController starts a new sync controller for a type config
@@ -117,14 +119,15 @@ func newKubeFedSyncController(controllerConfig *util.ControllerConfig, typeConfi
 	recorder := broadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: userAgent})
 
 	s := &KubeFedSyncController{
-		clusterAvailableDelay:   controllerConfig.ClusterAvailableDelay,
-		clusterUnavailableDelay: controllerConfig.ClusterUnavailableDelay,
-		smallDelay:              time.Second * 3,
-		eventRecorder:           recorder,
-		typeConfig:              typeConfig,
-		hostClusterClient:       client,
-		skipAdoptingResources:   controllerConfig.SkipAdoptingResources,
-		limitedScope:            controllerConfig.LimitedScope(),
+		clusterAvailableDelay:       controllerConfig.ClusterAvailableDelay,
+		clusterUnavailableDelay:     controllerConfig.ClusterUnavailableDelay,
+		smallDelay:                  time.Second * 3,
+		eventRecorder:               recorder,
+		typeConfig:                  typeConfig,
+		hostClusterClient:           client,
+		skipAdoptingResources:       controllerConfig.SkipAdoptingResources,
+		limitedScope:                controllerConfig.LimitedScope(),
+		rawResourceStatusCollection: controllerConfig.RawResourceStatusCollection,
 	}
 
 	s.worker = util.NewReconcileWorker(s.reconcile, util.WorkerTiming{
@@ -200,7 +203,7 @@ func (s *KubeFedSyncController) Run(stopChan <-chan struct{}) {
 // synced with the corresponding api server.
 func (s *KubeFedSyncController) isSynced() bool {
 	if !s.informer.ClustersSynced() {
-		klog.V(2).Infof("Cluster list not synced")
+		klog.V(2).Info("Cluster list not synced")
 		return false
 	}
 	if !s.fedAccessor.HasSynced() {
@@ -285,23 +288,27 @@ func (s *KubeFedSyncController) reconcile(qualifiedName util.QualifiedName) util
 // syncToClusters ensures that the state of the given object is
 // synchronized to member clusters.
 func (s *KubeFedSyncController) syncToClusters(fedResource FederatedResource) util.ReconciliationStatus {
+	// Enable raw resource status collection if the statusCollection is enabled for that type
+	// and the feature is also enabled.
+	enableRawResourceStatusCollection := s.typeConfig.GetStatusEnabled() && s.rawResourceStatusCollection
+
 	clusters, err := s.informer.GetClusters()
 	if err != nil {
 		fedResource.RecordError(string(status.ClusterRetrievalFailed), errors.Wrap(err, "Failed to retrieve list of clusters"))
-		return s.setFederatedStatus(fedResource, status.ClusterRetrievalFailed, nil)
+		return s.setFederatedStatus(fedResource, status.ClusterRetrievalFailed, nil, nil, enableRawResourceStatusCollection)
 	}
 
 	selectedClusterNames, err := fedResource.ComputePlacement(clusters)
 	if err != nil {
 		fedResource.RecordError(string(status.ComputePlacementFailed), errors.Wrap(err, "Failed to compute placement"))
-		return s.setFederatedStatus(fedResource, status.ComputePlacementFailed, nil)
+		return s.setFederatedStatus(fedResource, status.ComputePlacementFailed, nil, nil, enableRawResourceStatusCollection)
 	}
 
 	kind := fedResource.TargetKind()
 	key := fedResource.TargetName().String()
 	klog.V(4).Infof("Ensuring %s %q in clusters: %s", kind, key, strings.Join(selectedClusterNames.List(), ","))
 
-	dispatcher := dispatch.NewManagedDispatcher(s.informer.GetClientForCluster, fedResource, s.skipAdoptingResources)
+	dispatcher := dispatch.NewManagedDispatcher(s.informer.GetClientForCluster, fedResource, s.skipAdoptingResources, enableRawResourceStatusCollection)
 
 	for _, cluster := range clusters {
 		clusterName := cluster.Name
@@ -337,7 +344,7 @@ func (s *KubeFedSyncController) syncToClusters(fedResource FederatedResource) ut
 			}
 			if clusterObj.GetDeletionTimestamp() != nil {
 				// Resource is marked for deletion
-				dispatcher.RecordStatus(clusterName, status.WaitingForRemoval)
+				dispatcher.RecordStatus(clusterName, status.WaitingForRemoval, clusterObj.Object[util.StatusField])
 				continue
 			}
 			if fedResource.IsNamespaceInHostCluster(clusterObj) {
@@ -377,15 +384,19 @@ func (s *KubeFedSyncController) syncToClusters(fedResource FederatedResource) ut
 		runtime.HandleError(err)
 	}
 
-	collectedStatus := dispatcher.CollectedStatus()
-	return s.setFederatedStatus(fedResource, status.AggregateSuccess, &collectedStatus)
+	collectedStatus, collectedResourceStatus := dispatcher.CollectedStatus()
+	klog.V(4).Infof("Setting the federated status '%v'", collectedResourceStatus)
+	return s.setFederatedStatus(fedResource, status.AggregateSuccess, &collectedStatus, &collectedResourceStatus, enableRawResourceStatusCollection)
 }
 
 func (s *KubeFedSyncController) setFederatedStatus(fedResource FederatedResource,
-	reason status.AggregateReason, collectedStatus *status.CollectedPropagationStatus) util.ReconciliationStatus {
+	reason status.AggregateReason, collectedStatus *status.CollectedPropagationStatus, collectedResourceStatus *status.CollectedResourceStatus, resourceStatusCollection bool) util.ReconciliationStatus {
 	if collectedStatus == nil {
 		collectedStatus = &status.CollectedPropagationStatus{}
 	}
+	if collectedResourceStatus == nil {
+		collectedResourceStatus = &status.CollectedResourceStatus{}
+	}
 
 	kind := fedResource.FederatedKind()
 	name := fedResource.FederatedName()
@@ -405,13 +416,14 @@ func (s *KubeFedSyncController) setFederatedStatus(fedResource FederatedResource
 	// If the underlying resource has changed, attempt to retrieve and
 	// update it repeatedly.
 	err := wait.PollImmediate(1*time.Second, 5*time.Second, func() (bool, error) {
-		if updateRequired, err := status.SetFederatedStatus(obj, reason, *collectedStatus); err != nil {
+		if updateRequired, err := status.SetFederatedStatus(obj, reason, *collectedStatus, *collectedResourceStatus, resourceStatusCollection); err != nil {
+			klog.V(4).Infof("Failed to set the status for %s %q", kind, name)
 			return false, errors.Wrapf(err, "failed to set the status")
 		} else if !updateRequired {
 			klog.V(4).Infof("No status update necessary for %s %q", kind, name)
 			return true, nil
 		}
-
+		klog.V(4).Infof("Updating status for %s %q", kind, name)
 		err := s.hostClusterClient.UpdateStatus(context.TODO(), obj)
 		if err == nil {
 			return true, nil
diff --git a/pkg/controller/sync/dispatch/managed.go b/pkg/controller/sync/dispatch/managed.go
index 65f1160f..ec5f39af 100644
--- a/pkg/controller/sync/dispatch/managed.go
+++ b/pkg/controller/sync/dispatch/managed.go
@@ -30,6 +30,7 @@ import (
 	pkgruntime "k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/klog"
 
 	"sigs.k8s.io/kubefed/pkg/client/generic"
 	"sigs.k8s.io/kubefed/pkg/controller/sync/status"
@@ -60,10 +61,10 @@ type ManagedDispatcher interface {
 	Create(clusterName string)
 	Update(clusterName string, clusterObj *unstructured.Unstructured)
 	VersionMap() map[string]string
-	CollectedStatus() status.CollectedPropagationStatus
+	CollectedStatus() (status.CollectedPropagationStatus, status.CollectedResourceStatus)
 
 	RecordClusterError(propStatus status.PropagationStatus, clusterName string, err error)
-	RecordStatus(clusterName string, propStatus status.PropagationStatus)
+	RecordStatus(clusterName string, propStatus status.PropagationStatus, resourceStatus interface{})
 }
 
 type managedDispatcherImpl struct {
@@ -74,19 +75,24 @@ type managedDispatcherImpl struct {
 	fedResource           FederatedResourceForDispatch
 	versionMap            map[string]string
 	statusMap             status.PropagationStatusMap
+	resourceStatusMap     map[string]interface{}
 	skipAdoptingResources bool
 
 	// Track when resource updates are performed to allow indicating
 	// when a change was last propagated to member clusters.
 	resourcesUpdated bool
+
+	rawResourceStatusCollection bool
 }
 
-func NewManagedDispatcher(clientAccessor clientAccessorFunc, fedResource FederatedResourceForDispatch, skipAdoptingResources bool) ManagedDispatcher {
+func NewManagedDispatcher(clientAccessor clientAccessorFunc, fedResource FederatedResourceForDispatch, skipAdoptingResources, rawResourceStatusCollection bool) ManagedDispatcher {
 	d := &managedDispatcherImpl{
-		fedResource:           fedResource,
-		versionMap:            make(map[string]string),
-		statusMap:             make(status.PropagationStatusMap),
-		skipAdoptingResources: skipAdoptingResources,
+		fedResource:                 fedResource,
+		versionMap:                  make(map[string]string),
+		statusMap:                   make(status.PropagationStatusMap),
+		resourceStatusMap:           make(map[string]interface{}),
+		skipAdoptingResources:       skipAdoptingResources,
+		rawResourceStatusCollection: rawResourceStatusCollection,
 	}
 	d.dispatcher = newOperationDispatcher(clientAccessor, d)
 	d.unmanagedDispatcher = newUnmanagedDispatcher(d.dispatcher, d, fedResource.TargetGVK(), fedResource.TargetName())
@@ -132,7 +138,7 @@ func (d *managedDispatcherImpl) Create(clusterName string) {
 	// when a timeout occurs it won't be possible to determine which
 	// operation timed out.  The timeout status will be cleared by
 	// Wait() if a timeout does not occur.
-	d.RecordStatus(clusterName, status.CreationTimedOut)
+	d.RecordStatus(clusterName, status.CreationTimedOut, nil)
 	start := time.Now()
 	d.dispatcher.incrementOperationsInitiated()
 	const op = "create"
@@ -153,6 +159,7 @@ func (d *managedDispatcherImpl) Create(clusterName string) {
 		if err == nil {
 			version := util.ObjectVersion(obj)
 			d.recordVersion(clusterName, version)
+			d.RecordStatus(clusterName, status.CreationTimedOut, obj.Object[util.StatusField])
 			metrics.DispatchOperationDurationFromStart("create", start)
 			return util.StatusAllOK
 		}
@@ -172,6 +179,8 @@ func (d *managedDispatcherImpl) Create(clusterName string) {
 			return d.recordOperationError(status.RetrievalFailed, clusterName, op, wrappedErr)
 		}
 
+		d.RecordStatus(clusterName, status.CreationTimedOut, obj.Object[util.StatusField])
+
 		if d.skipAdoptingResources && !d.fedResource.IsNamespaceInHostCluster(obj) {
 			_ = d.recordOperationError(status.AlreadyExists, clusterName, op, errors.Errorf("Resource pre-exist in cluster"))
 			return util.StatusAllOK
@@ -185,7 +194,7 @@ func (d *managedDispatcherImpl) Create(clusterName string) {
 }
 
 func (d *managedDispatcherImpl) Update(clusterName string, clusterObj *unstructured.Unstructured) {
-	d.RecordStatus(clusterName, status.UpdateTimedOut)
+	d.RecordStatus(clusterName, status.UpdateTimedOut, clusterObj.Object[util.StatusField])
 
 	d.dispatcher.incrementOperationsInitiated()
 	const op = "update"
@@ -217,6 +226,7 @@ func (d *managedDispatcherImpl) Update(clusterName string, clusterObj *unstructu
 		}
 		if !util.ObjectNeedsUpdate(obj, clusterObj, version) {
 			// Resource is current
+			d.RecordStatus(clusterName, status.UpdateTimedOut, clusterObj.Object[util.StatusField])
 			return util.StatusAllOK
 		}
 
@@ -227,6 +237,7 @@ func (d *managedDispatcherImpl) Update(clusterName string, clusterObj *unstructu
 		if err != nil {
 			return d.recordOperationError(status.UpdateFailed, clusterName, op, err)
 		}
+		d.RecordStatus(clusterName, status.UpdateTimedOut, obj.Object[util.StatusField])
 		d.setResourcesUpdated()
 		version = util.ObjectVersion(obj)
 		d.recordVersion(clusterName, version)
@@ -235,31 +246,36 @@ func (d *managedDispatcherImpl) Update(clusterName string, clusterObj *unstructu
 }
 
 func (d *managedDispatcherImpl) Delete(clusterName string) {
-	d.RecordStatus(clusterName, status.DeletionTimedOut)
+	d.RecordStatus(clusterName, status.DeletionTimedOut, nil)
 
 	d.unmanagedDispatcher.Delete(clusterName)
 }
 
 func (d *managedDispatcherImpl) RemoveManagedLabel(clusterName string, clusterObj *unstructured.Unstructured) {
-	d.RecordStatus(clusterName, status.LabelRemovalTimedOut)
+	d.RecordStatus(clusterName, status.LabelRemovalTimedOut, clusterObj.Object[util.StatusField])
 
 	d.unmanagedDispatcher.RemoveManagedLabel(clusterName, clusterObj)
 }
 
 func (d *managedDispatcherImpl) RecordClusterError(propStatus status.PropagationStatus, clusterName string, err error) {
 	d.fedResource.RecordError(string(propStatus), err)
-	d.RecordStatus(clusterName, propStatus)
+	d.RecordStatus(clusterName, propStatus, nil)
 }
 
-func (d *managedDispatcherImpl) RecordStatus(clusterName string, propStatus status.PropagationStatus) {
+func (d *managedDispatcherImpl) RecordStatus(clusterName string, propStatus status.PropagationStatus, resourceStatus interface{}) {
 	d.Lock()
 	defer d.Unlock()
 	d.statusMap[clusterName] = propStatus
+
+	if d.rawResourceStatusCollection && resourceStatus != nil {
+		klog.V(4).Infof("Recording resource status %v", resourceStatus)
+		d.resourceStatusMap[clusterName] = resourceStatus
+	}
 }
 
 func (d *managedDispatcherImpl) recordOperationError(propStatus status.PropagationStatus, clusterName, operation string, err error) util.ReconciliationStatus {
 	d.recordError(clusterName, operation, err)
-	d.RecordStatus(clusterName, propStatus)
+	d.RecordStatus(clusterName, propStatus, nil)
 	return util.StatusError
 }
 
@@ -299,15 +315,23 @@ func (d *managedDispatcherImpl) setResourcesUpdated() {
 	d.resourcesUpdated = true
 }
 
-func (d *managedDispatcherImpl) CollectedStatus() status.CollectedPropagationStatus {
+func (d *managedDispatcherImpl) CollectedStatus() (status.CollectedPropagationStatus, status.CollectedResourceStatus) {
 	d.RLock()
 	defer d.RUnlock()
 	statusMap := make(status.PropagationStatusMap)
+	resourceStatusMap := make(map[string]interface{})
 	for key, value := range d.statusMap {
 		statusMap[key] = value
 	}
-	return status.CollectedPropagationStatus{
-		StatusMap:        statusMap,
-		ResourcesUpdated: d.resourcesUpdated,
+
+	for key, value := range d.resourceStatusMap {
+		resourceStatusMap[key] = value
 	}
+	return status.CollectedPropagationStatus{
+			StatusMap:        statusMap,
+			ResourcesUpdated: d.resourcesUpdated,
+		}, status.CollectedResourceStatus{
+			StatusMap:        resourceStatusMap,
+			ResourcesUpdated: d.resourcesUpdated,
+		}
 }
diff --git a/pkg/controller/sync/status/status.go b/pkg/controller/sync/status/status.go
index 5d7e980a..e8451510 100644
--- a/pkg/controller/sync/status/status.go
+++ b/pkg/controller/sync/status/status.go
@@ -18,6 +18,7 @@ package status
 
 import (
 	"encoding/json"
+	"reflect"
 	"time"
 
 	"github.com/pkg/errors"
@@ -25,6 +26,7 @@ import (
 	apiv1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/klog"
 
 	"sigs.k8s.io/kubefed/pkg/controller/util"
 )
@@ -71,8 +73,9 @@ const (
 )
 
 type GenericClusterStatus struct {
-	Name   string            `json:"name"`
-	Status PropagationStatus `json:"status,omitempty"`
+	Name         string            `json:"name"`
+	Status       PropagationStatus `json:"status,omitempty"`
+	RemoteStatus interface{}       `json:"remoteStatus,omitempty"`
 }
 
 type GenericCondition struct {
@@ -112,10 +115,15 @@ type CollectedPropagationStatus struct {
 	ResourcesUpdated bool
 }
 
+type CollectedResourceStatus struct {
+	StatusMap        map[string]interface{}
+	ResourcesUpdated bool
+}
+
 // SetFederatedStatus sets the conditions and clusters fields of the
 // federated resource's object map. Returns a boolean indication of
 // whether status should be written to the API.
-func SetFederatedStatus(fedObject *unstructured.Unstructured, reason AggregateReason, collectedStatus CollectedPropagationStatus) (bool, error) {
+func SetFederatedStatus(fedObject *unstructured.Unstructured, reason AggregateReason, collectedStatus CollectedPropagationStatus, collectedResourceStatus CollectedResourceStatus, resourceStatusCollection bool) (bool, error) {
 	resource := &GenericFederatedResource{}
 	err := util.UnstructuredToInterface(fedObject, resource)
 	if err != nil {
@@ -125,7 +133,7 @@ func SetFederatedStatus(fedObject *unstructured.Unstructured, reason AggregateRe
 		resource.Status = &GenericFederatedStatus{}
 	}
 
-	changed := resource.Status.update(fedObject.GetGeneration(), reason, collectedStatus)
+	changed := resource.Status.update(fedObject.GetGeneration(), reason, collectedStatus, collectedResourceStatus, resourceStatusCollection)
 	if !changed {
 		return false, nil
 	}
@@ -139,6 +147,8 @@ func SetFederatedStatus(fedObject *unstructured.Unstructured, reason AggregateRe
 	if err != nil {
 		return false, errors.Wrapf(err, "Failed to marshall generic resource json to unstructured")
 	}
+
+	klog.V(4).Infof("Setting the status of federated object '%v' and resource object '%v'", fedObject.GetName(), resourceObj.GetName())
 	fedObject.Object[util.StatusField] = resourceObj.Object[util.StatusField]
 
 	return true, nil
@@ -148,7 +158,7 @@ func SetFederatedStatus(fedObject *unstructured.Unstructured, reason AggregateRe
 // and collected status. Returns a boolean indication of whether the
 // status has been changed.
 func (s *GenericFederatedStatus) update(generation int64, reason AggregateReason,
-	collectedStatus CollectedPropagationStatus) bool {
+	collectedStatus CollectedPropagationStatus, collectedResourceStatus CollectedResourceStatus, resourceStatusCollection bool) bool {
 	generationUpdated := s.ObservedGeneration != generation
 	if generationUpdated {
 		s.ObservedGeneration = generation
@@ -157,54 +167,66 @@ func (s *GenericFederatedStatus) update(generation int64, reason AggregateReason
 	// Identify whether one or more clusters could not be reconciled
 	// successfully.
 	if reason == AggregateSuccess {
-		for _, value := range collectedStatus.StatusMap {
-			if value != ClusterPropagationOK {
+		for cluster, value := range collectedStatus.StatusMap {
+			rawStatus := collectedResourceStatus.StatusMap[cluster]
+			if value != ClusterPropagationOK || (resourceStatusCollection && rawStatus == nil) {
+				klog.V(4).Infof("Check the cluster '%v' with resource status '%v' and propStatus '%v' whose resource status collection is: '%v'", cluster, rawStatus, value, resourceStatusCollection)
 				reason = CheckClusters
 				break
 			}
 		}
 	}
 
-	clustersChanged := s.setClusters(collectedStatus.StatusMap)
+	clustersChanged := s.setClusters(collectedStatus.StatusMap, collectedResourceStatus.StatusMap)
 
 	// Indicate that changes were propagated if either status.clusters
 	// was changed or if existing resources were updated (which could
 	// occur even if status.clusters was unchanged).
-	changesPropagated := clustersChanged || len(collectedStatus.StatusMap) > 0 && collectedStatus.ResourcesUpdated
+	// TODO (hectorj2f): re-consider this new condition or add a new one for the resource status update or not.
+	changesPropagated := clustersChanged || len(collectedStatus.StatusMap) > 0 && len(collectedResourceStatus.StatusMap) > 0 && collectedStatus.ResourcesUpdated
 
 	propStatusUpdated := s.setPropagationCondition(reason, changesPropagated)
 
 	statusUpdated := generationUpdated || propStatusUpdated
+
+	klog.V(4).Infof("Value of flags: propStatusUpdated: '%v'; statusUpdated '%v'; changesPropagated '%v'", propStatusUpdated, statusUpdated, changesPropagated)
 	return statusUpdated
 }
 
-// setClusters sets the status.clusters slice from a propagation status
-// map. Returns a boolean indication of whether the status.clusters was
+// setClusters sets the status.clusters slice from propagation and resource status
+// maps. Returns a boolean indication of whether the status.clusters was
 // modified.
-func (s *GenericFederatedStatus) setClusters(statusMap PropagationStatusMap) bool {
-	if !s.clustersDiffers(statusMap) {
+func (s *GenericFederatedStatus) setClusters(statusMap PropagationStatusMap, resourceStatusMap map[string]interface{}) bool {
+	if !s.clustersDiffer(statusMap, resourceStatusMap) {
 		return false
 	}
 	s.Clusters = []GenericClusterStatus{}
 	for clusterName, status := range statusMap {
+		rawResourceStatus := resourceStatusMap[clusterName]
 		s.Clusters = append(s.Clusters, GenericClusterStatus{
-			Name:   clusterName,
-			Status: status,
+			Name:         clusterName,
+			Status:       status,
+			RemoteStatus: rawResourceStatus,
 		})
 	}
 	return true
 }
 
-// clustersDiffers checks whether `status.clusters` differs from the
+// clustersDiffer checks whether `status.clusters` differs from the
 // given status map.
-func (s *GenericFederatedStatus) clustersDiffers(statusMap PropagationStatusMap) bool {
-	if len(s.Clusters) != len(statusMap) {
+func (s *GenericFederatedStatus) clustersDiffer(statusMap PropagationStatusMap, resourceStatusMap map[string]interface{}) bool {
+	if len(s.Clusters) != len(statusMap) || len(s.Clusters) != len(resourceStatusMap) {
+		klog.V(4).Info("Clusters differs from the size")
 		return true
 	}
 	for _, status := range s.Clusters {
 		if statusMap[status.Name] != status.Status {
 			return true
 		}
+		if !reflect.DeepEqual(resourceStatusMap[status.Name], status.RemoteStatus) {
+			klog.V(4).Infof("Clusters resource status differ: %v VS %v", resourceStatusMap[status.Name], status.RemoteStatus)
+			return true
+		}
 	}
 	return false
 }
diff --git a/pkg/controller/util/controllerconfig.go b/pkg/controller/util/controllerconfig.go
index d0b5207a..af09f469 100644
--- a/pkg/controller/util/controllerconfig.go
+++ b/pkg/controller/util/controllerconfig.go
@@ -68,11 +68,12 @@ type ClusterHealthCheckConfig struct {
 // controllers.
 type ControllerConfig struct {
 	KubeFedNamespaces
-	KubeConfig              *restclient.Config
-	ClusterAvailableDelay   time.Duration
-	ClusterUnavailableDelay time.Duration
-	MinimizeLatency         bool
-	SkipAdoptingResources   bool
+	KubeConfig                  *restclient.Config
+	ClusterAvailableDelay       time.Duration
+	ClusterUnavailableDelay     time.Duration
+	MinimizeLatency             bool
+	SkipAdoptingResources       bool
+	RawResourceStatusCollection bool
 }
 
 func (c *ControllerConfig) LimitedScope() bool {
diff --git a/pkg/features/features.go b/pkg/features/features.go
index 98be20c2..d05b6eb3 100644
--- a/pkg/features/features.go
+++ b/pkg/features/features.go
@@ -53,6 +53,12 @@ const (
 	//
 	// DNS based federated ingress feature.
 	FederatedIngress featuregate.Feature = "FederatedIngress"
+
+	// owner: @hectorj2f
+	// beta: v0.5
+	//
+	// RawResourceStatusCollection enables the collection of the status of target types when enabled
+	RawResourceStatusCollection featuregate.Feature = "RawResourceStatusCollection"
 )
 
 func init() {
@@ -69,4 +75,5 @@ var DefaultKubeFedFeatureGates = map[featuregate.Feature]featuregate.FeatureSpec
 	PushReconciler:               {Default: true, PreRelease: featuregate.Beta},
 	CrossClusterServiceDiscovery: {Default: false, PreRelease: featuregate.Alpha},
 	FederatedIngress:             {Default: false, PreRelease: featuregate.Alpha},
+	RawResourceStatusCollection:  {Default: false, PreRelease: featuregate.Beta},
 }
diff --git a/pkg/kubefedctl/enable/validation.go b/pkg/kubefedctl/enable/validation.go
index 7c97b11a..5ee4e393 100644
--- a/pkg/kubefedctl/enable/validation.go
+++ b/pkg/kubefedctl/enable/validation.go
@@ -215,6 +215,10 @@ func ValidationSchema(specProps v1.JSONSchemaProps) *v1.CustomResourceValidation
 										"status": {
 											Type: "string",
 										},
+										"remoteStatus": {
+											XPreserveUnknownFields: pointer.BoolPtr(true),
+											Type:                   "object",
+										},
 									},
 									Required: []string{
 										"name",
diff --git a/scripts/create-clusters.sh b/scripts/create-clusters.sh
index 30a825fc..014ddc2c 100755
--- a/scripts/create-clusters.sh
+++ b/scripts/create-clusters.sh
@@ -25,7 +25,7 @@ set -o pipefail
 source "${BASH_SOURCE%/*}/util.sh"
 NUM_CLUSTERS="${NUM_CLUSTERS:-2}"
 KIND_IMAGE="${KIND_IMAGE:-}"
-KIND_TAG="${KIND_TAG:-v1.19.4@sha256:796d09e217d93bed01ecf8502633e48fd806fe42f9d02fdd468b81cd4e3bd40b}"
+KIND_TAG="${KIND_TAG:-}"
 OS="$(uname)"
 
 function create-clusters() {
diff --git a/scripts/deploy-kubefed.sh b/scripts/deploy-kubefed.sh
index ad565457..7f79f20a 100755
--- a/scripts/deploy-kubefed.sh
+++ b/scripts/deploy-kubefed.sh
@@ -80,7 +80,7 @@ function helm-deploy-cmd {
         --set controllermanager.webhook.repository=${repo} \
         --set controllermanager.webhook.image=${image} \
         --set controllermanager.webhook.tag=${tag} \
-        --set controllermanager.featureGates.CrossClusterServiceDiscovery=Enabled,controllermanager.featureGates.FederatedIngress=Enabled \
+        --set controllermanager.featureGates.CrossClusterServiceDiscovery=Enabled,controllermanager.featureGates.FederatedIngress=Enabled,controllermanager.featureGates.RawResourceStatusCollection=Enabled \
         ${force_redeploy_values:-} \
         --create-namespace \
         --wait"
@@ -173,10 +173,8 @@ cd -
 deploy-with-helm
 
 # Join the host cluster
-if [ -z "${NO_JOIN_HOST_CLUSTER:-}" ] ; then
-    CONTEXT="$(kubectl config current-context)"
-    ./bin/kubefedctl join "${CONTEXT}" --host-cluster-context "${CONTEXT}" --v=2 "${KF_NS_ARGS}" --error-on-existing=false
-fi
+CONTEXT="$(kubectl config current-context)"
+./bin/kubefedctl join "${CONTEXT}" --host-cluster-context "${CONTEXT}" --v=2 "${KF_NS_ARGS}" --error-on-existing=false
 
 for c in ${JOIN_CLUSTERS}; do
   ./bin/kubefedctl join "${c}" --host-cluster-context "${CONTEXT}" --v=2 "${KF_NS_ARGS}" --error-on-existing=false
diff --git a/scripts/pre-commit.sh b/scripts/pre-commit.sh
index 0632c567..899b4dd2 100755
--- a/scripts/pre-commit.sh
+++ b/scripts/pre-commit.sh
@@ -136,7 +136,7 @@ run-unit-tests
 echo "Downloading e2e test dependencies"
 ./scripts/download-e2e-binaries.sh
 
-KIND_TAG="v1.19.4@sha256:796d09e217d93bed01ecf8502633e48fd806fe42f9d02fdd468b81cd4e3bd40b" ./scripts/create-clusters.sh
+KIND_TAG="v1.19.1@sha256:98cf5288864662e37115e362b23e4369c8c4a408f99cbc06e58ac30ddc721600" ./scripts/create-clusters.sh
 
 declare -a join_cluster_list=() 
 if [[ -z "${JOIN_CLUSTERS}" ]]; then
diff --git a/scripts/sync-up-helm-chart.sh b/scripts/sync-up-helm-chart.sh
index 3d36ebed..490c44a9 100755
--- a/scripts/sync-up-helm-chart.sh
+++ b/scripts/sync-up-helm-chart.sh
@@ -78,11 +78,11 @@ users: []
 EOF
 
 # Start kube-apiserver to generate CRDs
-${ROOT_DIR}/bin/etcd --data-dir ${WORKDIR} --log-output stdout > ${WORKDIR}/etcd.log 2>&1 &
-util::wait-for-condition 'etcd' "curl http://127.0.0.1:2379/version &> /dev/null" 30
+${ROOT_DIR}/bin/etcd --data-dir ${WORKDIR} &
+util::wait-for-condition 'ok' "curl http://127.0.0.1:2379/version &> /dev/null" 30
 
-${ROOT_DIR}/bin/kube-apiserver --etcd-servers=http://127.0.0.1:2379 --service-cluster-ip-range=10.0.0.0/16 --cert-dir=${WORKDIR} --log-file=${WORKDIR}/kube-apiserver.log --logtostderr=false 2> /dev/null &
-util::wait-for-condition 'kube-apiserver' "kubectl --kubeconfig ${WORKDIR}/kubeconfig --context kubefed get --raw=/healthz &> /dev/null" 60
+${ROOT_DIR}/bin/kube-apiserver --etcd-servers=http://127.0.0.1:2379 --service-cluster-ip-range=10.0.0.0/16 --cert-dir ${WORKDIR} &
+util::wait-for-condition 'ok' "kubectl --kubeconfig ${WORKDIR}/kubeconfig --context kubefed get --raw=/healthz &> /dev/null" 60
 
 # Generate YAML templates to enable resource propagation for helm chart.
 echo -n > ${CHART_FEDERATED_PROPAGATION_DIR}/templates/federatedtypeconfig.yaml
@@ -104,4 +104,3 @@ done
 kill %1 # etcd
 kill %2 # kube-apiserver
 rm -fr ${WORKDIR}
-echo "Helm chart synced successfully"


Test patch:
diff --git a/pkg/controller/sync/status/status_test.go b/pkg/controller/sync/status/status_test.go
index 8f8b3136..7100a403 100644
--- a/pkg/controller/sync/status/status_test.go
+++ b/pkg/controller/sync/status/status_test.go
@@ -24,25 +24,43 @@ import (
 
 func TestGenericPropagationStatusUpdateChanged(t *testing.T) {
 	testCases := map[string]struct {
-		generation       int64
-		reason           AggregateReason
-		statusMap        PropagationStatusMap
-		resourcesUpdated bool
-		expectedChanged  bool
+		generation        int64
+		reason            AggregateReason
+		statusMap         PropagationStatusMap
+		resourceStatusMap map[string]interface{}
+		resourcesUpdated  bool
+		expectedChanged   bool
 	}{
 		"No change in clusters indicates unchanged": {
 			statusMap: PropagationStatusMap{
 				"cluster1": ClusterPropagationOK,
 			},
+			resourceStatusMap: map[string]interface{}{
+				"ready": false,
+				"stage": "absent",
+			},
+			resourcesUpdated: false,
+			expectedChanged:  true,
 		},
 		"No change in clusters with update indicates changed": {
 			statusMap: PropagationStatusMap{
 				"cluster1": ClusterPropagationOK,
 			},
+			resourceStatusMap: map[string]interface{}{
+				"ready": false,
+				"stage": "absent",
+			},
 			resourcesUpdated: true,
 			expectedChanged:  true,
 		},
 		"Change in clusters indicates changed": {
+			statusMap: PropagationStatusMap{
+				"cluster1": ClusterPropagationOK,
+			},
+			resourceStatusMap: map[string]interface{}{
+				"ready": true,
+				"stage": "deployed",
+			},
 			expectedChanged: true,
 		},
 		"Transition indicates changed": {
@@ -56,7 +74,7 @@ func TestGenericPropagationStatusUpdateChanged(t *testing.T) {
 	}
 	for testName, tc := range testCases {
 		t.Run(testName, func(t *testing.T) {
-			propStatus := &GenericFederatedStatus{
+			fedStatus := &GenericFederatedStatus{
 				Clusters: []GenericClusterStatus{
 					{
 						Name: "cluster1",
@@ -73,7 +91,11 @@ func TestGenericPropagationStatusUpdateChanged(t *testing.T) {
 				StatusMap:        tc.statusMap,
 				ResourcesUpdated: tc.resourcesUpdated,
 			}
-			changed := propStatus.update(tc.generation, tc.reason, collectedStatus)
+			collectedResourceStatus := CollectedResourceStatus{
+				StatusMap:        tc.resourceStatusMap,
+				ResourcesUpdated: tc.resourcesUpdated,
+			}
+			changed := fedStatus.update(tc.generation, tc.reason, collectedStatus, collectedResourceStatus, true)
 			if tc.expectedChanged != changed {
 				t.Fatalf("Expected changed to be %v, got %v", tc.expectedChanged, changed)
 			}
diff --git a/test/common/bindata.go b/test/common/bindata.go
index d10da883..a3685ac5 100644
--- a/test/common/bindata.go
+++ b/test/common/bindata.go
@@ -390,6 +390,8 @@ spec:
   featureGates:
   - name: PushReconciler
     configuration: "Enabled"
+  - name: RawResourceStatusCollection
+    configuration: "Enabled"
   - name: SchedulerPreferences
     configuration: "Enabled"
   - name: CrossClusterServiceDiscovery
diff --git a/test/common/crudtester.go b/test/common/crudtester.go
index e2fb2d1f..83d493c6 100644
--- a/test/common/crudtester.go
+++ b/test/common/crudtester.go
@@ -32,6 +32,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
+	kubeclientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 
 	"sigs.k8s.io/kubefed/pkg/apis/core/common"
@@ -714,27 +715,129 @@ func (c *FederatedTypeCrudTester) versionForCluster(version *fedv1a1.PropagatedV
 	return ""
 }
 
+func (c *FederatedTypeCrudTester) CheckRemoteStatus(fedObject *unstructured.Unstructured, targetObject *unstructured.Unstructured) {
+	for clusterName := range c.testClusters {
+		clusterConfig := c.testClusters[clusterName].Config
+
+		kubeClient := kubeclientset.NewForConfigOrDie(clusterConfig)
+		WaitForNamespaceOrDie(c.tl, kubeClient, clusterName, targetObject.GetNamespace(),
+			c.waitInterval, 30*time.Second)
+
+		util.AddManagedLabel(targetObject)
+		labeledObj, err := CreateResource(clusterConfig, c.typeConfig.GetTargetType(), targetObject)
+		if err != nil {
+			c.tl.Fatalf("Failed to create labeled resource in cluster %q: %v", clusterName, err)
+		}
+
+		clusterClient := genericclient.NewForConfigOrDie(clusterConfig)
+
+		c.tl.Log("Checking that the resource has status")
+		var objStatus interface{}
+		err = wait.PollImmediate(c.waitInterval, wait.ForeverTestTimeout, func() (bool, error) {
+			obj := &unstructured.Unstructured{}
+			obj.SetGroupVersionKind(labeledObj.GroupVersionKind())
+			err := clusterClient.Get(context.TODO(), obj, labeledObj.GetNamespace(), labeledObj.GetName())
+			if err != nil {
+				c.tl.Errorf("Error retrieving kubefed cluster object resource: %v", err)
+				return false, nil
+			}
+
+			objStatus = obj.Object[util.StatusField]
+			c.tl.Logf("Show kubefed cluster object status: %v", objStatus)
+			return (obj.Object[util.StatusField] != nil), nil
+		})
+		if err != nil {
+			c.tl.Fatal("Timed out waiting for the resource to have a status field")
+		}
+		c.tl.Logf("Kubefed cluster object status: %v", objStatus)
+
+		c.tl.Log("Checking that the federated resource has a remote status field")
+		objRemoteStatus, err := c.getRemoteStatus(fedObject, clusterName)
+		if err != nil {
+			c.tl.Fatal("Timed out waiting for the federated resource to have a remote status field")
+		}
+		if objRemoteStatus == nil {
+			c.tl.Fatal("Federated object remote status is empty")
+		}
+		c.tl.Logf("Show federated object remote status %v", objRemoteStatus)
+
+		err = clusterClient.Delete(context.TODO(), labeledObj, labeledObj.GetNamespace(), labeledObj.GetName())
+		if err != nil {
+			c.tl.Fatalf("Unexpected error deleting the labeled resource: %v", err)
+		}
+	}
+}
+
+func (c *FederatedTypeCrudTester) getRemoteStatus(fedObject *unstructured.Unstructured, clusterName string) (interface{}, error) {
+	apiResource := c.typeConfig.GetFederatedType()
+	qualifiedName := util.NewQualifiedName(fedObject)
+
+	client := c.resourceClient(apiResource)
+	var remoteStatusObj interface{}
+	// The default is normally 30 seconds
+	waitTimeout := 6 * wait.ForeverTestTimeout
+	err := wait.PollImmediate(c.waitInterval, waitTimeout, func() (bool, error) {
+		fedObj, err := client.Resources(qualifiedName.Namespace).Get(context.Background(), qualifiedName.Name, metav1.GetOptions{})
+		if err != nil {
+			c.tl.Errorf("An unexpected error occurred while polling for remote status: %v", err)
+			return false, nil
+		}
+
+		resource := &status.GenericFederatedResource{}
+		err = util.UnstructuredToInterface(fedObj, resource)
+		if err != nil {
+			return false, err
+		}
+		if resource.Status != nil {
+			for _, cluster := range resource.Status.Clusters {
+				c.tl.Logf("Current status of resource for cluster '%s' with value: %v", cluster.Name, resource.Status)
+				if cluster.Name == clusterName && cluster.Status == status.ClusterPropagationOK {
+					c.tl.Logf("resource remote status for cluster '%s': %v", cluster.Name, cluster.RemoteStatus)
+					if cluster.RemoteStatus != nil {
+						remoteStatusObj = cluster.RemoteStatus
+						return true, nil
+					}
+				}
+			}
+		}
+		return false, nil
+	})
+
+	if err != nil {
+		c.tl.Fatalf("Timed out waiting for the remote status: %q", qualifiedName)
+		return nil, err
+	}
+
+	return remoteStatusObj, nil
+}
+
 func (c *FederatedTypeCrudTester) CheckStatusCreated(qualifiedName util.QualifiedName) {
 	if !c.typeConfig.GetStatusEnabled() {
 		return
 	}
 
 	statusAPIResource := c.typeConfig.GetStatusType()
-	statusKind := statusAPIResource.Kind
+	// TODO (hectorj2f): To be deprecated.
+	// I added a check to support current FederatedServiceStatus and new remote resource
+	// status. The new mechanism reuses the statusCollection flag but it does not define
+	// a StatusType type.
+	if statusAPIResource != nil {
+		statusKind := statusAPIResource.Kind
 
-	c.tl.Logf("Checking creation of %s %q", statusKind, qualifiedName)
+		c.tl.Logf("Checking creation of %s %q", statusKind, qualifiedName)
 
-	client := c.resourceClient(*statusAPIResource)
-	err := wait.PollImmediate(c.waitInterval, wait.ForeverTestTimeout, func() (bool, error) {
-		_, err := client.Resources(qualifiedName.Namespace).Get(context.Background(), qualifiedName.Name, metav1.GetOptions{})
-		if err != nil && !apierrors.IsNotFound(err) {
-			c.tl.Errorf("An unexpected error occurred while polling for desired status: %v", err)
-		}
-		return (err == nil), nil
-	})
+		client := c.resourceClient(*statusAPIResource)
+		err := wait.PollImmediate(c.waitInterval, wait.ForeverTestTimeout, func() (bool, error) {
+			_, err := client.Resources(qualifiedName.Namespace).Get(context.Background(), qualifiedName.Name, metav1.GetOptions{})
+			if err != nil && !apierrors.IsNotFound(err) {
+				c.tl.Errorf("An unexpected error occurred while polling for desired status: %v", err)
+			}
+			return (err == nil), nil
+		})
 
-	if err != nil {
-		c.tl.Fatalf("Timed out waiting for %s %q", statusKind, qualifiedName)
+		if err != nil {
+			c.tl.Fatalf("Timed out waiting for %s %q", statusKind, qualifiedName)
+		}
 	}
 }
 
diff --git a/test/common/typeconfig.go b/test/common/typeconfig.go
index 23479a5d..c6334b82 100644
--- a/test/common/typeconfig.go
+++ b/test/common/typeconfig.go
@@ -33,3 +33,12 @@ func GetTypeConfig(genericClient client.Client, name, namespace string) (typecon
 
 	return typeConfig, nil
 }
+
+func UpdateTypeConfig(genericClient client.Client, typeConfig *fedv1b1.FederatedTypeConfig, namespace string) error {
+	err := genericClient.Update(context.Background(), typeConfig)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
diff --git a/test/e2e/crud.go b/test/e2e/crud.go
index e51752e2..1c6d6710 100644
--- a/test/e2e/crud.go
+++ b/test/e2e/crud.go
@@ -29,6 +29,7 @@ import (
 	kubeclientset "k8s.io/client-go/kubernetes"
 
 	"sigs.k8s.io/kubefed/pkg/apis/core/typeconfig"
+	"sigs.k8s.io/kubefed/pkg/apis/core/v1beta1"
 	genericclient "sigs.k8s.io/kubefed/pkg/client/generic"
 	"sigs.k8s.io/kubefed/pkg/controller/sync/status"
 	"sigs.k8s.io/kubefed/pkg/controller/util"
@@ -39,6 +40,8 @@ import (
 	. "github.com/onsi/ginkgo" //nolint:stylecheck
 )
 
+var containedTypeNames = []string{"jobs.batch", "deployments.apps", "replicasets.apps"}
+
 type testObjectsAccessor func(namespace string, clusterNames []string) (targetObject *unstructured.Unstructured, overrides []interface{}, err error)
 
 var _ = Describe("Federated", func() {
@@ -62,6 +65,33 @@ var _ = Describe("Federated", func() {
 				crudTester.CheckLifecycle(targetObject, overrides)
 			})
 
+			for _, remoteStatusTypeName := range containedTypeNames {
+				if typeConfigName == remoteStatusTypeName {
+
+					It("should be created, read its remote status and deleted successfully", func() {
+						kubeFedConfig := &v1beta1.KubeFedConfig{}
+						client := genericclient.NewForConfigOrDie(f.KubeConfig())
+						err := client.Get(context.TODO(), kubeFedConfig, f.KubeFedSystemNamespace(), util.KubeFedConfigName)
+						if err != nil {
+							tl.Fatalf("Error collecting the kubefedconfig file: %v", err)
+						}
+						tl.Logf("Show the content of the kubefedconfig file: '%v'", kubeFedConfig)
+
+						typeConfig, testObjectsFunc := getCrudTestInput(f, tl, typeConfigName, fixture)
+						crudTester, targetObject, overrides := initCrudTest(f, tl, typeConfig, testObjectsFunc)
+						fedObject := crudTester.CheckCreate(targetObject, overrides)
+
+						By("Checking the remote status filled for each federated resource for every cluster")
+						tl.Logf("Checking the existence of a remote status for each fedObj in every cluster: %v", fedObject)
+						crudTester.CheckRemoteStatus(fedObject, targetObject)
+
+						defer func() {
+							crudTester.CheckDelete(fedObject, false)
+						}()
+					})
+				}
+			}
+
 			// The tests that follow only need to be executed against
 			// a single namespaced type.
 			if typeConfigName != "configmaps" {
@@ -248,6 +278,21 @@ func getCrudTestInput(f framework.KubeFedFramework, tl common.TestLogger,
 		tl.Fatalf("Error retrieving federatedtypeconfig %q: %v", typeConfigName, err)
 	}
 
+	// Enable the Status Collection for this type of resource
+	tc := typeConfig.(*v1beta1.FederatedTypeConfig)
+	for _, typeName := range containedTypeNames {
+		tl.Logf("TypeConfig name: %s", tc.GetName())
+		if tc.GetName() == typeName {
+			tl.Logf("Enabling remote status collection for %v", typeConfig.GetFederatedType().Kind)
+			statusCollection := v1beta1.StatusCollectionEnabled
+			tc.Spec.StatusCollection = &statusCollection
+			err = common.UpdateTypeConfig(client, tc, f.KubeFedSystemNamespace())
+			if err != nil {
+				tl.Fatalf("Error updating the federatedtypeconfig %q: %v", typeConfigName, err)
+			}
+		}
+	}
+
 	if framework.TestContext.LimitedScope && !typeConfig.GetNamespaced() {
 		framework.Skipf("Federation of cluster-scoped type %s is not supported by a namespaced control plane.", typeConfigName)
 	}
@@ -262,6 +307,7 @@ func getCrudTestInput(f framework.KubeFedFramework, tl common.TestLogger,
 			targetObject.SetName(namespace)
 			targetObject.SetNamespace(namespace)
 		}
+
 		overrides, err := common.OverridesFromFixture(clusterNames, fixture)
 		if err != nil {
 			return nil, nil, err
@@ -297,6 +343,7 @@ func initCrudTestWithPropagation(f framework.KubeFedFramework, tl common.TestLog
 
 	kubeConfig := f.KubeConfig()
 	targetAPIResource := typeConfig.GetTargetType()
+
 	testClusters := f.ClusterDynamicClients(&targetAPIResource, userAgent)
 	crudTester, err := common.NewFederatedTypeCrudTester(tl, typeConfig, kubeConfig, testClusters, framework.PollInterval, framework.TestContext.SingleCallTimeout)
 	if err != nil {
diff --git a/test/e2e/defaulting.go b/test/e2e/defaulting.go
index dff779cc..acdac36c 100644
--- a/test/e2e/defaulting.go
+++ b/test/e2e/defaulting.go
@@ -88,6 +88,10 @@ var _ = Describe("Default", func() {
 							Name:          string(features.PushReconciler),
 							Configuration: v1beta1.ConfigurationDisabled,
 						},
+						{
+							Name:          string(features.RawResourceStatusCollection),
+							Configuration: v1beta1.ConfigurationEnabled,
+						},
 					},
 					ClusterHealthCheck: &v1beta1.ClusterHealthCheckConfig{
 
diff --git a/test/e2e/framework/controller.go b/test/e2e/framework/controller.go
index ae1a3943..3f3db32b 100644
--- a/test/e2e/framework/controller.go
+++ b/test/e2e/framework/controller.go
@@ -50,7 +50,13 @@ func NewSyncControllerFixture(tl common.TestLogger, controllerConfig *util.Contr
 	if err != nil {
 		tl.Fatalf("Error starting sync controller: %v", err)
 	}
-	if typeConfig.GetStatusEnabled() {
+	// The status controller should be only enabled for services whenever the statusAPIResource is defined
+	if typeConfig.GetStatusEnabled() && typeConfig.GetTargetType().Kind == "Service" {
+		federatedAPIResource := typeConfig.GetFederatedType()
+		statusAPIResource := typeConfig.GetStatusType()
+		if statusAPIResource == nil {
+			tl.Fatalf("Skipping status collection, status API resource is not defined %q", federatedAPIResource.Kind)
+		}
 		err := status.StartKubeFedStatusController(controllerConfig, f.stopChan, typeConfig)
 		if err != nil {
 			tl.Fatalf("Error starting status controller: %v", err)
diff --git a/test/e2e/framework/unmanaged.go b/test/e2e/framework/unmanaged.go
index f35b633c..133398e4 100644
--- a/test/e2e/framework/unmanaged.go
+++ b/test/e2e/framework/unmanaged.go
@@ -69,6 +69,7 @@ func SetUpFeatureGates() {
 		string(features.PushReconciler):               true,
 		string(features.CrossClusterServiceDiscovery): true,
 		string(features.FederatedIngress):             true,
+		string(features.RawResourceStatusCollection):  true,
 	}
 	err := utilfeature.DefaultMutableFeatureGate.SetFromMap(resetDefaultFeatureGates)
 	Expect(err).NotTo(HaveOccurred())
@@ -86,7 +87,8 @@ func SetUpControlPlane() {
 		KubeFedNamespaces: util.KubeFedNamespaces{
 			KubeFedNamespace: TestContext.KubeFedSystemNamespace,
 		},
-		KubeConfig: config,
+		KubeConfig:                  config,
+		RawResourceStatusCollection: true,
 	})
 }
 
@@ -183,7 +185,7 @@ func (f *UnmanagedFramework) AfterEach() {
 }
 
 func (f *UnmanagedFramework) ControllerConfig() *util.ControllerConfig {
-	return &util.ControllerConfig{
+	controllerCfg := &util.ControllerConfig{
 		KubeFedNamespaces: util.KubeFedNamespaces{
 			KubeFedNamespace: TestContext.KubeFedSystemNamespace,
 			TargetNamespace:  f.inMemoryTargetNamespace(),
@@ -191,6 +193,8 @@ func (f *UnmanagedFramework) ControllerConfig() *util.ControllerConfig {
 		KubeConfig:      f.Config,
 		MinimizeLatency: true,
 	}
+	controllerCfg.RawResourceStatusCollection = true
+	return controllerCfg
 }
 
 func (f *UnmanagedFramework) Logger() common.TestLogger {