From 1b32d84d52ab10f1452601e3c906d65aa85071fd Mon Sep 17 00:00:00 2001
From: lijianguo <lijianguo1991@outlook.com>
Date: Tue, 19 Aug 2025 16:48:40 +0800
Subject: [PATCH] feat(blocklist): blocklist api

---
 .../templates/controller-clusterrole.yaml     |   6 +
 rootfs/api/backend.py                         |   2 +-
 .../0025_alter_app_id_alter_gateway_name.py   |  25 ++
 .../migrations/0026_app_suspended_state.py    |  18 ++
 rootfs/api/models/app.py                      |   1 +
 rootfs/api/models/blocklist.py                | 231 ++++++++++++++++++
 rootfs/api/tasks.py                           |  16 ++
 rootfs/api/views.py                           |  21 +-
 rootfs/dev_requirements.txt                   |   3 +
 rootfs/scheduler/mock.py                      | 158 +++++++++++-
 rootfs/scheduler/resources/daemonset.py       |  45 ++++
 rootfs/scheduler/resources/statefulset.py     |  45 ++++
 rootfs/scheduler/tests/test_daemonsets.py     |  72 ++++++
 rootfs/scheduler/tests/test_statefulsets.py   |  57 +++++
 14 files changed, 689 insertions(+), 11 deletions(-)
 create mode 100644 rootfs/api/migrations/0025_alter_app_id_alter_gateway_name.py
 create mode 100644 rootfs/api/migrations/0026_app_suspended_state.py
 create mode 100644 rootfs/scheduler/resources/daemonset.py
 create mode 100644 rootfs/scheduler/resources/statefulset.py
 create mode 100644 rootfs/scheduler/tests/test_daemonsets.py
 create mode 100644 rootfs/scheduler/tests/test_statefulsets.py

diff --git a/charts/controller/templates/controller-clusterrole.yaml b/charts/controller/templates/controller-clusterrole.yaml
index 611129a13..3809127e3 100644
--- a/charts/controller/templates/controller-clusterrole.yaml
+++ b/charts/controller/templates/controller-clusterrole.yaml
@@ -51,6 +51,12 @@ rules:
 - apiGroups: ["extensions", "apps"]
   resources: ["deployments/scale", "replicasets/scale"]
   verbs: ["get", "update"]
+- apiGroups: ["apps"]
+  resources: ["statefulsets"]
+  verbs: ["get", "list", "patch"]
+- apiGroups: ["extensions", "apps"]
+  resources: ["daemonsets"]
+  verbs: ["get", "list", "patch"]
 - apiGroups: ["extensions", "autoscaling"]
   resources: ["horizontalpodautoscalers"]
   verbs: ["get", "list", "create", "update", "delete"]
diff --git a/rootfs/api/backend.py b/rootfs/api/backend.py
index f629f15a6..f5108fe3d 100644
--- a/rootfs/api/backend.py
+++ b/rootfs/api/backend.py
@@ -21,7 +21,7 @@ class DryccOIDC(OpenIdConnectAuth):
     USERINFO_URL = settings.SOCIAL_AUTH_DRYCC_USERINFO_URL
     JWKS_URI = settings.SOCIAL_AUTH_DRYCC_JWKS_URI
     OIDC_ENDPOINT = settings.SOCIAL_AUTH_DRYCC_OIDC_ENDPOINT
-    DEFAULT_SCOPE = ['openid']
+    DEFAULT_SCOPE = ['openid', 'profile', 'email']
     EXTRA_DATA = [
         ('id', 'id'),
         ('access_token', 'access_token'),
diff --git a/rootfs/api/migrations/0025_alter_app_id_alter_gateway_name.py b/rootfs/api/migrations/0025_alter_app_id_alter_gateway_name.py
new file mode 100644
index 000000000..eeb228016
--- /dev/null
+++ b/rootfs/api/migrations/0025_alter_app_id_alter_gateway_name.py
@@ -0,0 +1,25 @@
+# Generated by Django 4.2.22 on 2025-08-22 02:48
+
+import api.models.app
+import api.utils
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api', '0024_limitplan_container_volume_mounts_and_more'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='app',
+            name='id',
+            field=models.SlugField(max_length=63, null=True, unique=True, validators=[api.models.app.validate_app_id]),
+        ),
+        migrations.AlterField(
+            model_name='gateway',
+            name='name',
+            field=models.CharField(db_index=True, max_length=63, validators=[api.utils.validate_label]),
+        ),
+    ]
diff --git a/rootfs/api/migrations/0026_app_suspended_state.py b/rootfs/api/migrations/0026_app_suspended_state.py
new file mode 100644
index 000000000..a667b809f
--- /dev/null
+++ b/rootfs/api/migrations/0026_app_suspended_state.py
@@ -0,0 +1,18 @@
+# Generated by Django 4.2.22 on 2025-08-22 02:56
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api', '0025_alter_app_id_alter_gateway_name'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='app',
+            name='suspended_state',
+            field=models.JSONField(blank=True, default=dict),
+        ),
+    ]
diff --git a/rootfs/api/models/app.py b/rootfs/api/models/app.py
index 3e921b69e..e286dda62 100644
--- a/rootfs/api/models/app.py
+++ b/rootfs/api/models/app.py
@@ -114,6 +114,7 @@ class App(UuidAuditedModel):
                           validators=[validate_app_id])
     structure = models.JSONField(
         default=dict, blank=True, validators=[validate_app_structure])
+    suspended_state = models.JSONField(default=dict, blank=True)
 
     class Meta:
         verbose_name = 'Application'
diff --git a/rootfs/api/models/blocklist.py b/rootfs/api/models/blocklist.py
index 5f5f6eac3..306aa4d09 100644
--- a/rootfs/api/models/blocklist.py
+++ b/rootfs/api/models/blocklist.py
@@ -1,9 +1,16 @@
+import logging
+import functools
 from django.db import models
 from django.contrib.auth import get_user_model
+from scheduler import KubeHTTPException
+from api.exceptions import ServiceUnavailable
+from api.utils import apply_tasks
 from .app import App
 from .base import UuidAuditedModel
 
+
 User = get_user_model()
+logger = logging.getLogger(__name__)
 
 
 class Blocklist(UuidAuditedModel):
@@ -41,3 +48,227 @@ def get_blocklist(cls, app: App):
     class Meta:
         ordering = ['-created']
         unique_together = (("id", "type"),)
+
+    def related_resource_deployments(self, app: App):
+        "get resource deployments"
+        try:
+            deployments = self.scheduler.deployment.get(app.id, labels={}).json()['items']  # noqa
+            if not deployments:
+                deployments = []
+            data = []
+            for d in deployments:
+                item = {
+                    'name': d['metadata']['name'],
+                    'replicas': d['spec'].get("replicas", 0),
+                }
+                data.append(item)
+            data.sort(key=lambda x: x['name'])
+            return data
+        except KubeHTTPException:
+            pass
+        except Exception as e:
+            err = f'(list resource deployments): {e}'
+            logger.info(err)
+            raise ServiceUnavailable(err) from e
+
+    def related_resource_statefulsets(self, app: App):
+        "get resource statefulsets"
+        try:
+            statefulsets = self.scheduler.statefulset.get(app.id, labels={}).json()['items']  # noqa
+            if not statefulsets:
+                statefulsets = []
+            data = []
+            for s in statefulsets:
+                item = {
+                    'name': s['metadata']['name'],
+                    'replicas': s['spec'].get("replicas", 0),
+                }
+                data.append(item)
+            data.sort(key=lambda x: x['name'])
+            return data
+        except KubeHTTPException:
+            pass
+        except Exception as e:
+            err = f'(list resource statefulsets): {e}'
+            logger.info(err)
+            raise ServiceUnavailable(err) from e
+
+    def related_resource_daemonsets(self, app: App):
+        "get resource daemonsets"
+        try:
+            daemonsets = self.scheduler.daemonset.get(app.id, labels={}).json()['items']  # noqa
+            if not daemonsets:
+                daemonsets = []
+            data = []
+            for d in daemonsets:
+                item = {
+                    'name': d['metadata']['name'],
+                    'affinity': d['spec']['template']['spec'].get('affinity', {}),
+                }
+                data.append(item)
+            data.sort(key=lambda x: x['name'])
+            return data
+        except KubeHTTPException:
+            pass
+        except Exception as e:
+            err = f'(list resource daemonsets): {e}'
+            logger.info(err)
+            raise ServiceUnavailable(err) from e
+
+    def suspended_state(self, app: App):
+        """
+        Get deployments/statefulsets/daemonsets with labels app.kubernetes.io/managed-by=Helm
+        Store in the suspended_state field of the app model
+        Format:
+        {"deployments": [{"name":"sample1", "replicas": 1}],
+        "statefulsets": [{"name":"sample1", "replicas": 1}],
+        "daemonsets": [{"name":"sample1", "affinity": {}}]}
+        """
+        suspended_state = {}
+        suspended_state['deployments'] = self.related_resource_deployments(app)
+        suspended_state['statefulsets'] = self.related_resource_statefulsets(app)
+        suspended_state['daemonsets'] = self.related_resource_daemonsets(app)
+        return suspended_state
+
+    def scale_resource_deployments(self, app: App, deployment_name: str, replicas=0):
+        """scale deployments"""
+        try:
+            deployment = self.scheduler.deployment.get(app.id, deployment_name).json()
+            self.scheduler.scales.update(app.id, deployment_name, replicas, deployment)
+        except KubeHTTPException:
+            pass
+        except Exception as e:
+            err = f'(scale resource deployments): {e}'
+            logger.info(err)
+            raise ServiceUnavailable(err) from e
+
+    def scale_resource_statefulsets(self, app: App, statefulset_name: str, replicas=0):
+        """scale statefulsets"""
+        try:
+            self.scheduler.statefulset.get(app.id, statefulset_name).json()
+            manifest = {
+                'spec': {
+                    'persistentVolumeClaimRetentionPolicy': {
+                        'whenScaled': 'Retain'
+                    },
+                    'replicas': replicas
+                }
+            }
+            self.scheduler.statefulset.patch(app.id, statefulset_name, manifest)
+        except KubeHTTPException:
+            pass
+        except Exception as e:
+            err = f'(scale resource statefulsets): {e}'
+            logger.info(err)
+            raise ServiceUnavailable(err) from e
+
+    def scale_resource_daemonsets(self, app: App, daemonset_name: str, manifest: dict):
+        """set affinity"""
+        try:
+            self.scheduler.daemonset.get(app.id, daemonset_name).json()
+            self.scheduler.daemonset.patch(app.id, daemonset_name, manifest)
+        except KubeHTTPException:
+            pass
+        except Exception as e:
+            err = f'(scale resource daemonsets): {e}'
+            logger.info(err)
+            raise ServiceUnavailable(err) from e
+
+    def scale_resources(self, app: App, suspended_state: dict, scale_type="block"):
+        "scale resources tasks"
+        tasks = []
+        deployments = suspended_state.get('deployments', [])
+        if not deployments:
+            deployments = []
+        for d in deployments:
+            if scale_type == "unblock":
+                replicas = d['replicas']
+            else:
+                replicas = 0
+            tasks.append((
+                functools.partial(
+                    self.scale_resource_deployments,
+                    app=app,
+                    deployment_name=d['name'],
+                    replicas=replicas
+                ),
+                lambda future, name=d["name"]: app.log(
+                    f'{scale_type} scale deployment {name} callback: {future.result()}',
+                )
+            ))
+
+        statefulsets = suspended_state.get('statefulsets', [])
+        if not statefulsets:
+            statefulsets = []
+        for s in statefulsets:
+            if scale_type == "unblock":
+                replicas = s['replicas']
+            else:
+                replicas = 0
+            tasks.append((
+                functools.partial(
+                    self.scale_resource_statefulsets,
+                    app=app,
+                    statefulset_name=s['name'],
+                    replicas=replicas
+                ),
+                lambda future, name=s["name"]: app.log(
+                    f'{scale_type} scale statefulset {name} callback: {future.result()}',
+                )
+            ))
+
+        daemonsets = suspended_state.get('daemonsets', [])
+        if not daemonsets:
+            daemonsets = []
+        for d in daemonsets:
+            if scale_type == "unblock":
+                manifest = {
+                    "spec": {
+                        "template": {
+                            "spec": {
+                                "affinity": d['affinity']
+                            }
+                        }
+                    }
+                }
+            else:
+                manifest = {
+                    "spec": {
+                        "template": {
+                            "spec": {
+                                "affinity": {
+                                    "nodeAffinity": {
+                                        "requiredDuringSchedulingIgnoredDuringExecution": {
+                                            "nodeSelectorTerms": [{
+                                                "matchExpressions": [{
+                                                    "key": "kubernetes.io/hostname",
+                                                    "operator": "In",
+                                                    "values": [
+                                                        "nohostname"
+                                                    ]
+                                                }]
+                                            }]
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            tasks.append((
+                functools.partial(
+                    self.scale_resource_daemonsets,
+                    app=app,
+                    daemonset_name=d['name'],
+                    manifest=manifest
+                ),
+                lambda future, name=d["name"]: app.log(
+                    f'{scale_type} scale daemonset {name} callback: {future.result()}',
+                )
+            ))
+        try:
+            apply_tasks(tasks)
+        except Exception as e:
+            err = f'({scale_type} scale resources): {e}'
+            logger.info(err)
+            raise ServiceUnavailable(err) from e
diff --git a/rootfs/api/tasks.py b/rootfs/api/tasks.py
index 4591dba72..9132d6f93 100644
--- a/rootfs/api/tasks.py
+++ b/rootfs/api/tasks.py
@@ -158,3 +158,19 @@ def downstream_model_owner(app, old_owner, new_owner):
         raise e
     else:
         signals.request_finished.send(sender=task_id)
+
+
+@shared_task(
+    autoretry_for=(ServiceUnavailable, ),
+    retry_kwargs={'max_retries': None}
+)
+def scale_resources(blocklist, app, suspended_state, scale_type):
+    task_id = uuid.uuid4().hex
+    signals.request_started.send(sender=task_id)
+    try:
+        blocklist.scale_resources(app, suspended_state, scale_type)
+    except Exception as e:
+        signals.got_request_exception.send(sender=task_id)
+        raise e
+    else:
+        signals.request_finished.send(sender=task_id)
diff --git a/rootfs/api/views.py b/rootfs/api/views.py
index c6d738ff1..de9be4abf 100644
--- a/rootfs/api/views.py
+++ b/rootfs/api/views.py
@@ -35,7 +35,7 @@
 
 from api import monitor, models, permissions, serializers, viewsets, authentication, __version__
 from api.tasks import scale_app, restart_app, mount_app, downstream_model_owner, \
-    delete_pod
+    delete_pod, scale_resources
 from api.exceptions import AlreadyExists, ServiceUnavailable, DryccException
 
 from django.views.decorators.cache import never_cache
@@ -183,13 +183,17 @@ class WorkflowManagerViewset(GenericViewSet):
 
     def block(self, request,  **kwargs):
         try:
-            blocklist, _ = models.blocklist.Blocklist.objects.get_or_create(
+            blocklist, created = models.blocklist.Blocklist.objects.get_or_create(
                 id=kwargs['id'],
                 type=models.blocklist.Blocklist.get_type(kwargs["type"]),
                 defaults={"remark": request.data.get("remark")}
             )
             for app in blocklist.related_apps:
-                scale_app.delay(app, app.owner, {key: 0 for key in app.structure.keys()})
+                if created:
+                    app.suspended_state = blocklist.suspended_state(app)
+                    app.save()
+                # scale to 0
+                scale_resources.delay(blocklist, app, app.suspended_state, "block")
             return HttpResponse(status=201)
         except ValueError as e:
             logger.info(e)
@@ -197,10 +201,17 @@ def block(self, request,  **kwargs):
 
     def unblock(self, request,  **kwargs):
         try:
-            models.blocklist.Blocklist.objects.filter(
+            blocklists = models.blocklist.Blocklist.objects.filter(
                 id=kwargs['id'],
                 type=models.blocklist.Blocklist.get_type(kwargs["type"])
-            ).delete()
+            )
+            for blocklist in blocklists:
+                for app in blocklist.related_apps:
+                    # scale to up
+                    scale_resources.delay(blocklist, app, app.suspended_state, "unblock")
+                    app.suspended_state = {}
+                    app.save()
+                blocklist.delete()
             return HttpResponse(status=204)
         except ValueError as e:
             logger.info(e)
diff --git a/rootfs/dev_requirements.txt b/rootfs/dev_requirements.txt
index 9f542ed42..1654cf51a 100644
--- a/rootfs/dev_requirements.txt
+++ b/rootfs/dev_requirements.txt
@@ -8,3 +8,6 @@ flake8==7.2.0
 
 # mock out python-requests, mostly k8s
 requests-mock==1.12.1
+
+# tracebacks
+tblib==3.1.0
\ No newline at end of file
diff --git a/rootfs/scheduler/mock.py b/rootfs/scheduler/mock.py
index 6d1ac5e62..7d016475e 100644
--- a/rootfs/scheduler/mock.py
+++ b/rootfs/scheduler/mock.py
@@ -18,6 +18,8 @@
 from django.conf import settings
 from django.core.cache import cache
 
+from api import utils
+
 import logging
 logger = logging.getLogger(__name__)
 
@@ -84,6 +86,7 @@ def _acquire(self):
     'persistentvolumeclaims', 'serviceinstances', 'servicebindings',
     'limitranges', 'gateways', 'httproutes', 'tcproutes', 'udproutes',
     'issuers', 'certificates', 'certificaterequests', 'jobs', 'clusterserviceclasses',
+    'statefulsets', 'daemonsets',
 ]
 
 
@@ -902,7 +905,7 @@ def patch(request, context):  # noqa: C901
     data = request.json()
 
     # merge new data into old but keep labels separate in case they changed
-    if 'labels' in data['metadata']:
+    if 'labels' in data.get('metadata', {}):
         labels = data['metadata'].pop('labels')
         item['metadata'].update(data['metadata'])
         data['metadata'] = item['metadata']
@@ -935,6 +938,10 @@ def patch(request, context):  # noqa: C901
     elif resource_type in ['gateways']:
         data['status'] = item['status']
         cache.set(url, data, None)
+    elif resource_type in ['daemonsets', 'statefulsets']:
+        # item data merge-patch
+        data = utils.dict_merge(item, data)
+        cache.set(url, data, None)
     else:
         # Update the individual resource
         change_cache(url, resource_type, data)
@@ -1106,7 +1113,14 @@ def __init__(self, url, k8s_api_verify_tls=True, metadata=None):
             self.ns.get('duplicate')
         except KubeHTTPException:
             self.ns.create('duplicate')
-
+        # node
+        self._setup_node()
+        # daemonset
+        self._setup_daemonset()
+        # statefulset
+        self._setup_statefulset()
+
+    def _setup_node(self):
         try:
             self.node.get('172.17.8.100')
         except KubeHTTPException:
@@ -1183,12 +1197,146 @@ def __init__(self, url, k8s_api_verify_tls=True, metadata=None):
                     }
                 }
             }
-
             name = 'api_v1_nodes_172.17.8.100'
             cache.set(name, data, None)
             cache.set('api_v1_nodes', [name], None)
-        except Exception as e:
-            logger.critical(e)
+
+    def _setup_daemonset(self):
+        data = {
+            "apiVersion": "apps/v1",
+            "kind": "DaemonSet",
+            "metadata": {
+                "name": "drycc",
+                "namespace": "drycc",
+                "labels": {
+                    "app.kubernetes.io/managed-by": "Helm",
+                    "app.kubernetes.io/name": "drycc"
+                }
+            },
+            "spec": {
+                "selector": {
+                    "matchLabels": {
+                        "app.kubernetes.io/name": "drycc"
+                    }
+                },
+                "template": {
+                    "metadata": {
+                        "labels": {
+                            "app.kubernetes.io/managed-by": "Helm",
+                            "app.kubernetes.io/name": "drycc"
+                        }
+                    },
+                    "spec": {
+                        "affinity": {
+                            "nodeAffinity": {
+                                "requiredDuringSchedulingIgnoredDuringExecution": {
+                                    "nodeSelectorTerms": [
+                                        {
+                                            "matchExpressions": [{
+                                                "key": "kubernetes.io/hostname",
+                                                "operator": "In",
+                                                "values": [
+                                                    "server-01",
+                                                    "server-02",
+                                                    "server-03"
+                                                ]}
+                                            ]
+                                        }
+                                    ]
+                                }
+                            }
+                        },
+                        "containers": [{
+                            "image": "registry.drycc.cc/drycc/drycc",
+                            "imagePullPolicy": "IfNotPresent",
+                            "name": "drycc"
+                        }]
+                    }
+                }
+            },
+            "status": {
+                "currentNumberScheduled": 3,
+                "desiredNumberScheduled": 3,
+                "numberAvailable": 3,
+                "numberMisscheduled": 0,
+                "numberReady": 3,
+                "observedGeneration": 2,
+                "updatedNumberScheduled": 3
+            }
+        }
+        name = 'apis_apps_v1_namespaces_drycc_daemonsets_drycc'
+        cache.set(name, data, None)
+        cache.set('apis_apps_v1_namespaces_drycc_daemonsets', [name], None)
+
+    def _setup_statefulset(self):
+        data = {
+            "apiVersion": "apps/v1",
+            "kind": "StatefulSet",
+            "metadata": {
+                "labels": {
+                    "app.kubernetes.io/managed-by": "Helm",
+                    "app.kubernetes.io/name": "drycc"
+                },
+                "name": "drycc",
+                "namespace": "drycc"
+            },
+            "spec": {
+                "persistentVolumeClaimRetentionPolicy": {
+                    "whenDeleted": "Delete",
+                    "whenScaled": "Retain"
+                },
+                "selector": {
+                    "matchLabels": {
+                        "app.kubernetes.io/name": "drycc"
+                    }
+                },
+                "template": {
+                    "metadata": {
+                        "labels": {
+                            "app.kubernetes.io/managed-by": "Helm",
+                            "app.kubernetes.io/name": "drycc"
+                        }
+                    },
+                    "spec": {
+                            "containers": [{
+                                "image": "registry.drycc.cc/drycc/drycc",
+                                "imagePullPolicy": "IfNotPresent",
+                                "name": "drycc"
+                            }]
+                        }
+                    },
+                "volumeClaimTemplates": [{
+                    "apiVersion": "v1",
+                    "kind": "PersistentVolumeClaim",
+                    "metadata": {
+                        "name": "work-data"
+                    },
+                    "spec": {
+                        "accessModes": [
+                            "ReadWriteOnce"
+                        ],
+                        "resources": {
+                            "requests": {
+                                "storage": "1Gi"
+                            }
+                        },
+                        "volumeMode": "Filesystem"
+                    }
+                }]
+            },
+            "status": {
+                "availableReplicas": 2,
+                "collisionCount": 0,
+                "currentReplicas": 2,
+                "observedGeneration": 4,
+                "readyReplicas": 2,
+                "replicas": 2,
+                "updatedReplicas": 2
+            }
+        }
+        name = 'apis_apps_v1_namespaces_drycc_statefulsets_drycc'
+        cache.set(name, data, None)
+        cache.set('apis_apps_v1_namespaces_drycc_statefulsets', [name], None)
 
 
 scheduler.session = session()
diff --git a/rootfs/scheduler/resources/daemonset.py b/rootfs/scheduler/resources/daemonset.py
new file mode 100644
index 000000000..9d7dfbd33
--- /dev/null
+++ b/rootfs/scheduler/resources/daemonset.py
@@ -0,0 +1,45 @@
+import json
+import logging
+from scheduler.resources import Resource
+from scheduler.exceptions import KubeHTTPException
+
+
+class Daemonset(Resource):
+    api_prefix = 'apis'
+    api_version = 'apps/v1'
+
+    def get(self, namespace, name=None, ignore_exception=False, **kwargs):
+        """
+        Fetch a single Daemonsets or a list
+        """
+        url = '/namespaces/{}/daemonsets'
+        args = [namespace]
+        if name is not None:
+            args.append(name)
+            url += '/{}'
+            message = 'get Daemonset "{}" in Namespace "{}"'
+        else:
+            message = 'get Daemonsets in Namespace "{}"'
+
+        url = self.api(url, *args)
+        response = self.http_get(url, params=self.query_params(**kwargs))
+        if self.unhealthy(response.status_code) and not ignore_exception:
+            args.reverse()  # error msg is in reverse order
+            raise KubeHTTPException(response, message, *args)
+
+        return response
+
+    def patch(self, namespace, name, manifest, ignore_exception=False, **kwargs):
+        url = self.api("/namespaces/{}/daemonsets/{}", namespace, name)
+        response = self.http_patch(
+            url,
+            json=manifest,
+            headers={"Content-Type": "application/merge-patch+json"}
+        )
+
+        if self.unhealthy(response.status_code):
+            self.log(
+                namespace, 'template: {}'.format(json.dumps(manifest, indent=4)), logging.DEBUG)
+            if not ignore_exception:
+                raise KubeHTTPException(response, 'patch Daemonset "{}"', name)
+        return response
diff --git a/rootfs/scheduler/resources/statefulset.py b/rootfs/scheduler/resources/statefulset.py
new file mode 100644
index 000000000..e7b0cff8d
--- /dev/null
+++ b/rootfs/scheduler/resources/statefulset.py
@@ -0,0 +1,45 @@
+import json
+import logging
+from scheduler.resources import Resource
+from scheduler.exceptions import KubeHTTPException
+
+
+class Statefulset(Resource):
+    api_prefix = 'apis'
+    api_version = 'apps/v1'
+
+    def get(self, namespace, name=None, ignore_exception=False, **kwargs):
+        """
+        Fetch a single Statefulsets or a list
+        """
+        url = '/namespaces/{}/statefulsets'
+        args = [namespace]
+        if name is not None:
+            args.append(name)
+            url += '/{}'
+            message = 'get Statefulset "{}" in Namespace "{}"'
+        else:
+            message = 'get Statefulsets in Namespace "{}"'
+
+        url = self.api(url, *args)
+        response = self.http_get(url, params=self.query_params(**kwargs))
+        if self.unhealthy(response.status_code) and not ignore_exception:
+            args.reverse()  # error msg is in reverse order
+            raise KubeHTTPException(response, message, *args)
+
+        return response
+
+    def patch(self, namespace, name, manifest, ignore_exception=False, **kwargs):
+        url = self.api("/namespaces/{}/statefulsets/{}", namespace, name)
+        response = self.http_patch(
+            url,
+            json=manifest,
+            headers={"Content-Type": "application/merge-patch+json"}
+        )
+
+        if self.unhealthy(response.status_code):
+            self.log(
+                namespace, 'template: {}'.format(json.dumps(manifest, indent=4)), logging.DEBUG)
+            if not ignore_exception:
+                raise KubeHTTPException(response, 'patch Statefulset "{}"', name)
+        return response
diff --git a/rootfs/scheduler/tests/test_daemonsets.py b/rootfs/scheduler/tests/test_daemonsets.py
new file mode 100644
index 000000000..f4c8ab985
--- /dev/null
+++ b/rootfs/scheduler/tests/test_daemonsets.py
@@ -0,0 +1,72 @@
+"""
+Unit tests for the Drycc scheduler module.
+
+Run the tests with "./manage.py test scheduler"
+"""
+from scheduler.tests import TestCase
+from scheduler import KubeHTTPException
+
+
+class DaemonsetsTest(TestCase):
+    """Tests scheduler daemonset calls"""
+
+    def test_get_daemonsets(self):
+        response = self.scheduler.daemonset.get('drycc')
+        data = response.json()
+        self.assertEqual(response.status_code, 200, data)
+        self.assertIn('items', data)
+        self.assertEqual(1, len(data['items']))
+        # simple verify of data
+        self.assertEqual(data['items'][0]['metadata']['name'], 'drycc')
+
+    def test_get_daemonset(self):
+        with self.assertRaises(
+            KubeHTTPException,
+            msg='failed to get Daemonset doesnotexist in Namespace drycc: 404 Not Found'  # noqa
+        ):
+            self.scheduler.daemonset.get('drycc', 'doesnotexist')
+        namespace = 'drycc'
+        name = 'drycc'
+        response = self.scheduler.daemonset.get(namespace, name)
+        data = response.json()
+        self.assertEqual(response.status_code, 200, data)
+        self.assertEqual(data['apiVersion'], 'apps/v1')
+        self.assertEqual(data['kind'], 'DaemonSet')
+        self.assertEqual(data['metadata']['name'], name)
+
+    def test_patch_daemonset(self):
+        namespace = 'drycc'
+        name = 'drycc'
+        manifest_affinity = {
+            "nodeAffinity": {
+                "requiredDuringSchedulingIgnoredDuringExecution": {
+                    "nodeSelectorTerms": [{
+                        "matchExpressions": [{
+                            "key": "kubernetes.io/hostname",
+                            "operator": "In",
+                            "values": [
+                                "nohostname"
+                            ]
+                        }]
+                    }]
+                }
+            }
+        }
+        response = self.scheduler.daemonset.patch(
+            namespace,
+            name,
+            manifest={
+                "spec": {
+                    "template": {
+                        "spec": {
+                            "affinity": manifest_affinity
+                        }
+                    }
+                }
+            },
+        )
+        affinity = response.json()["spec"]["template"]['spec']['affinity']
+        self.assertEqual(
+            affinity,
+            manifest_affinity
+        )
diff --git a/rootfs/scheduler/tests/test_statefulsets.py b/rootfs/scheduler/tests/test_statefulsets.py
new file mode 100644
index 000000000..292b017a0
--- /dev/null
+++ b/rootfs/scheduler/tests/test_statefulsets.py
@@ -0,0 +1,57 @@
+"""
+Unit tests for the Drycc scheduler module.
+
+Run the tests with "./manage.py test scheduler"
+"""
+from scheduler.tests import TestCase
+from scheduler import KubeHTTPException
+
+
+class StatefulsetsTest(TestCase):
+    """Tests scheduler statefulset calls"""
+
+    def test_get_statefulsets(self):
+        response = self.scheduler.statefulset.get('drycc')
+        data = response.json()
+        self.assertEqual(response.status_code, 200, data)
+        self.assertIn('items', data)
+        self.assertEqual(1, len(data['items']))
+        # simple verify of data
+        self.assertEqual(data['items'][0]['metadata']['name'], 'drycc')
+
+    def test_get_statefulset(self):
+        with self.assertRaises(
+            KubeHTTPException,
+            msg='failed to get Statefulset doesnotexist in Namespace drycc: 404 Not Found'  # noqa
+        ):
+            self.scheduler.statefulset.get('drycc', 'doesnotexist')
+        namespace = 'drycc'
+        name = 'drycc'
+        response = self.scheduler.statefulset.get(namespace, name)
+        data = response.json()
+        self.assertEqual(response.status_code, 200, data)
+        self.assertEqual(data['apiVersion'], 'apps/v1')
+        self.assertEqual(data['kind'], 'StatefulSet')
+        self.assertEqual(data['metadata']['name'], name)
+
+    def test_patch_statefulset(self):
+        namespace = 'drycc'
+        name = 'drycc'
+        manifest = {
+            'spec': {
+                'persistentVolumeClaimRetentionPolicy': {
+                    'whenScaled': 'Retain'
+                },
+                'replicas': 0
+            }
+        }
+        response = self.scheduler.statefulset.patch(
+            namespace,
+            name,
+            manifest=manifest,
+        )
+        when_scaled = response.json()["spec"]["persistentVolumeClaimRetentionPolicy"]['whenScaled']
+        self.assertEqual(
+            when_scaled,
+            'Retain'
+        )
