diff --git a/fiaas_deploy_daemon/__init__.py b/fiaas_deploy_daemon/__init__.py
index 96a5392a..7698dab3 100644
--- a/fiaas_deploy_daemon/__init__.py
+++ b/fiaas_deploy_daemon/__init__.py
@@ -205,7 +205,7 @@ def main():
     try:
         log.info("fiaas-deploy-daemon starting with configuration {!r}".format(cfg))
         if cfg.enable_crd_support:
-            crd_binding = CustomResourceDefinitionBindings(cfg.use_apiextensionsv1_crd)
+            crd_binding = CustomResourceDefinitionBindings(cfg.use_apiextensionsv1_crd, cfg.include_status_in_app)
         else:
             crd_binding = DisabledCustomResourceDefinitionBindings()
         binding_specs = [
diff --git a/fiaas_deploy_daemon/bootstrap/bootstrapper.py b/fiaas_deploy_daemon/bootstrap/bootstrapper.py
index c7d8368d..9ae4e3cc 100644
--- a/fiaas_deploy_daemon/bootstrap/bootstrapper.py
+++ b/fiaas_deploy_daemon/bootstrap/bootstrapper.py
@@ -71,7 +71,7 @@ def __init__(self, config, deploy_queue, spec_factory, lifecycle):
             from ..crd.status import connect_signals
         else:
             raise InvalidConfigurationException("Custom Resource Definition support must be enabled when bootstrapping")
-        connect_signals()
+        connect_signals(config.include_status_in_app)
         signal(DEPLOY_STATUS_CHANGED).connect(self._store_status)
 
     def run(self):
diff --git a/fiaas_deploy_daemon/config.py b/fiaas_deploy_daemon/config.py
index deb85a14..37166160 100644
--- a/fiaas_deploy_daemon/config.py
+++ b/fiaas_deploy_daemon/config.py
@@ -256,7 +256,7 @@ def _parse_args(self, args):
         parser.add_argument(
             "--deployment-max-surge",
             help="maximum number of extra pods that can be scheduled above the desired "
-            "number of pods during an update",
+                 "number of pods during an update",
             default="25%",
             type=_int_or_unicode,
         )
@@ -271,7 +271,7 @@ def _parse_args(self, args):
             "--ready-check-timeout-multiplier",
             type=int,
             help="Multiply default ready check timeout (replicas * initialDelaySeconds) with this "
-            + "number of seconds  (default: %(default)s)",
+                 + "number of seconds  (default: %(default)s)",
             default=10,
         )
         parser.add_argument(
@@ -302,6 +302,10 @@ def _parse_args(self, args):
             action="store_true",
             default=False,
         )
+        parser.add_argument(
+            "--include-status-in-app", help="Include status subresource in application CRD", default=False,
+            action="store_true"
+        )
         usage_reporting_parser = parser.add_argument_group("Usage Reporting", USAGE_REPORTING_LONG_HELP)
         usage_reporting_parser.add_argument(
             "--usage-reporting-cluster-name", help="Name of the cluster where the fiaas-deploy-daemon instance resides"
diff --git a/fiaas_deploy_daemon/crd/__init__.py b/fiaas_deploy_daemon/crd/__init__.py
index 9dc14bd0..2a75146b 100644
--- a/fiaas_deploy_daemon/crd/__init__.py
+++ b/fiaas_deploy_daemon/crd/__init__.py
@@ -26,8 +26,9 @@
 
 
 class CustomResourceDefinitionBindings(pinject.BindingSpec):
-    def __init__(self, use_apiextensionsv1_crd):
+    def __init__(self, use_apiextensionsv1_crd, include_status_in_app):
         self.use_apiextensionsv1_crd = use_apiextensionsv1_crd
+        self.include_status_in_app = include_status_in_app
 
     def configure(self, bind, require):
         require("config")
@@ -38,7 +39,7 @@ def configure(self, bind, require):
             bind("crd_resources_syncer", to_class=CrdResourcesSyncerApiextensionsV1)
         else:
             bind("crd_resources_syncer", to_class=CrdResourcesSyncerApiextensionsV1Beta1)
-        connect_signals()
+        connect_signals(self.include_status_in_app)
 
 
 class DisabledCustomResourceDefinitionBindings(pinject.BindingSpec):
diff --git a/fiaas_deploy_daemon/crd/crd_resources_syncer_apiextensionsv1.py b/fiaas_deploy_daemon/crd/crd_resources_syncer_apiextensionsv1.py
index f60194a6..572cb0b4 100644
--- a/fiaas_deploy_daemon/crd/crd_resources_syncer_apiextensionsv1.py
+++ b/fiaas_deploy_daemon/crd/crd_resources_syncer_apiextensionsv1.py
@@ -26,6 +26,9 @@
     CustomResourceDefinitionVersion,
     CustomResourceValidation,
     JSONSchemaProps,
+    CustomResourceSubresources,
+    JSONSchemaPropsStatusEnabled,
+    CustomResourceSubresourceStatusEnabled
 )
 
 from ..retry import retry_on_upsert_conflict
@@ -36,12 +39,13 @@
 class CrdResourcesSyncerApiextensionsV1(object):
     @staticmethod
     @retry_on_upsert_conflict
-    def _create_or_update(kind, plural, short_names, group, schema_properties):
+    def _create_or_update(kind, plural, short_names, group, open_apiv3_schema, subresources=None):
         name = "%s.%s" % (plural, group)
         metadata = ObjectMeta(name=name)
         names = CustomResourceDefinitionNames(kind=kind, plural=plural, shortNames=short_names)
-        schema = CustomResourceValidation(openAPIV3Schema=JSONSchemaProps(type="object", properties=schema_properties))
-        version_v1 = CustomResourceDefinitionVersion(name="v1", served=True, storage=True, schema=schema)
+        schema = CustomResourceValidation(openAPIV3Schema=open_apiv3_schema)
+        version_v1 = CustomResourceDefinitionVersion(name="v1", served=True, storage=True, schema=schema,
+                                                     subresources=subresources)
         spec = CustomResourceDefinitionSpec(
             group=group,
             names=names,
@@ -94,7 +98,27 @@ def update_crd_resources(cls):
                             "status": object_with_unknown_fields,
                         },
                     },
-                },
+                }
+            },
+            "status": {
+                "type": "object",
+                "properties": {
+                    "result": {
+                        "type": "string"
+                    },
+                    "observedGeneration": {
+                        "type": "integer"
+                    },
+                    "logs": {
+                        "type": "array",
+                        "items": {
+                            "type": "string"
+                        }
+                    },
+                    "deployment_id": {
+                        "type": "string"
+                    }
+                }
             }
         }
         application_status_schema_properties = {
@@ -102,12 +126,14 @@ def update_crd_resources(cls):
             "logs": {"type": "array", "items": {"type": "string"}},
         }
         cls._create_or_update(
-            "Application", "applications", ("app", "fa"), "fiaas.schibsted.io", application_schema_properties
+            "Application", "applications", ("app", "fa"), "fiaas.schibsted.io",
+            JSONSchemaPropsStatusEnabled(type="object", properties=application_schema_properties),
+            CustomResourceSubresources(status=CustomResourceSubresourceStatusEnabled())
         )
         cls._create_or_update(
             "ApplicationStatus",
             "application-statuses",
             ("status", "appstatus", "fs"),
             "fiaas.schibsted.io",
-            application_status_schema_properties,
+            JSONSchemaProps(type="object", properties=application_status_schema_properties)
         )
diff --git a/fiaas_deploy_daemon/crd/status.py b/fiaas_deploy_daemon/crd/status.py
index aab05573..a5274828 100644
--- a/fiaas_deploy_daemon/crd/status.py
+++ b/fiaas_deploy_daemon/crd/status.py
@@ -24,9 +24,9 @@
 from k8s.client import NotFound
 from k8s.models.common import ObjectMeta, OwnerReference
 
-from .types import FiaasApplicationStatus
+from .types import FiaasApplication, FiaasApplicationStatus, FiaasApplicationStatusResult
 from ..lifecycle import DEPLOY_STATUS_CHANGED, STATUS_STARTED
-from ..log_extras import get_final_logs, get_running_logs
+from ..log_extras import get_final_error_logs, get_final_logs, get_running_error_logs, get_running_logs
 from ..retry import retry_on_upsert_conflict
 from ..tools import merge_dicts
 
@@ -35,8 +35,11 @@
 LOG = logging.getLogger(__name__)
 
 
-def connect_signals():
-    signal(DEPLOY_STATUS_CHANGED).connect(_handle_signal)
+def connect_signals(include_status_in_app):
+    if include_status_in_app:
+        signal(DEPLOY_STATUS_CHANGED).connect(_handle_signal_with_status)
+    else:
+        signal(DEPLOY_STATUS_CHANGED).connect(_handle_signal_without_status)
 
 
 def now():
@@ -45,16 +48,49 @@ def now():
     return now.isoformat()
 
 
-def _handle_signal(sender, status, subject):
+def _handle_signal_without_status(sender, status, subject):
+    if status == STATUS_STARTED:
+        status = "RUNNING"
+    else:
+        status = status.upper()
+
+    _save_status(status, subject)
+    _cleanup(subject.app_name, subject.namespace)
+
+
+def _handle_signal_with_status(sender, status, subject):
     if status == STATUS_STARTED:
         status = "RUNNING"
     else:
         status = status.upper()
 
+    _save_status_inline(status, subject)
     _save_status(status, subject)
     _cleanup(subject.app_name, subject.namespace)
 
 
+@retry_on_upsert_conflict
+def _save_status_inline(result, subject):
+    (uid, app_name, namespace, deployment_id, repository, labels, annotations) = subject
+
+    app = FiaasApplication.get(app_name, namespace)
+    generation = int(app.metadata.generation)
+    try:
+        application_deployment_id = app.metadata.labels["fiaas/deployment_id"]
+    except (AttributeError, KeyError, TypeError):
+        raise ValueError("The Application {} is missing the 'fiaas/deployment_id' label".format(app_name))
+    # We only want to get error logs here.
+    if deployment_id == application_deployment_id:
+        logs = _get_error_logs(app_name, namespace, deployment_id, result)
+
+        LOG.info("Saving inline result %s for %s/%s deployment_id=%s generation %s", result, namespace, app_name, deployment_id, generation)
+        app.status = FiaasApplicationStatusResult(observedGeneration=generation, result=result, logs=logs,
+                                                  deployment_id=deployment_id)
+        app.save_status()
+    else:
+        LOG.debug("Skipping saving status for application %s with different deployment_id", app_name)
+
+
 @retry_on_upsert_conflict
 def _save_status(result, subject):
     (uid, app_name, namespace, deployment_id, repository, labels, annotations) = subject
@@ -97,6 +133,11 @@ def _get_logs(app_name, namespace, deployment_id, result):
     )
 
 
+def _get_error_logs(app_name, namespace, deployment_id, result):
+    return get_running_error_logs(app_name, namespace, deployment_id) if result in [u"RUNNING", u"INITIATED"] else \
+       get_final_error_logs(app_name, namespace, deployment_id)
+
+
 def _cleanup(app_name=None, namespace=None):
     statuses = FiaasApplicationStatus.find(app_name, namespace)
 
diff --git a/fiaas_deploy_daemon/crd/types.py b/fiaas_deploy_daemon/crd/types.py
index 3ef8f7da..e580bf8a 100644
--- a/fiaas_deploy_daemon/crd/types.py
+++ b/fiaas_deploy_daemon/crd/types.py
@@ -41,6 +41,13 @@ class FiaasApplicationSpec(Model):
     additional_annotations = Field(AdditionalLabelsOrAnnotations)
 
 
+class FiaasApplicationStatusResult(Model):
+    result = RequiredField(six.text_type)
+    observedGeneration = Field(int, 0)  # NOQA
+    logs = ListField(six.text_type)
+    deployment_id = Field(six.text_type)
+
+
 class FiaasApplication(Model):
     class Meta:
         list_url = "/apis/fiaas.schibsted.io/v1/applications"
@@ -54,6 +61,7 @@ class Meta:
 
     metadata = Field(ObjectMeta)
     spec = Field(FiaasApplicationSpec)
+    status = Field(FiaasApplicationStatusResult)
 
 
 class FiaasApplicationStatus(Model):
diff --git a/fiaas_deploy_daemon/crd/watcher.py b/fiaas_deploy_daemon/crd/watcher.py
index 195d6b0b..2f1975f1 100644
--- a/fiaas_deploy_daemon/crd/watcher.py
+++ b/fiaas_deploy_daemon/crd/watcher.py
@@ -68,6 +68,19 @@ def _handle_watch_event(self, event):
         else:
             raise ValueError("Unknown WatchEvent type {}".format(event.type))
 
+    # When we receive update event on FiaasApplication
+    # don't deploy if it's a status update
+    def _skip_status_event(self, application):
+        app_name = application.spec.application
+        deployment_id = application.metadata.labels["fiaas/deployment_id"]
+        generation = int(application.metadata.generation)
+        observed_generation = int(application.status.observedGeneration)
+        deployment_id_status = application.status.deployment_id
+        if observed_generation == generation and deployment_id == deployment_id_status:
+            LOG.debug("Skipping watch event created from status update %s for app %s", deployment_id, app_name)
+            return True
+        return False
+
     def _deploy(self, application):
         app_name = application.spec.application
         LOG.debug("Deploying %s", app_name)
@@ -76,6 +89,8 @@ def _deploy(self, application):
             set_extras(app_name=app_name, namespace=application.metadata.namespace, deployment_id=deployment_id)
         except (AttributeError, KeyError, TypeError):
             raise ValueError("The Application {} is missing the 'fiaas/deployment_id' label".format(app_name))
+        if self._skip_status_event(application):
+            return
         if self._already_deployed(app_name, application.metadata.namespace, deployment_id):
             LOG.debug("Have already deployed %s for app %s", deployment_id, app_name)
             return
diff --git a/fiaas_deploy_daemon/log_extras.py b/fiaas_deploy_daemon/log_extras.py
index 1c63f8d3..ff1a69eb 100644
--- a/fiaas_deploy_daemon/log_extras.py
+++ b/fiaas_deploy_daemon/log_extras.py
@@ -20,6 +20,7 @@
 from collections import defaultdict
 
 _LOGS = defaultdict(list)
+_ERROR_LOGS = defaultdict(list)
 _LOG_EXTRAS = threading.local()
 _LOG_FORMAT = (
     "[%(asctime)s|%(levelname)7s] %(message)s " "[%(name)s|%(threadName)s|%(extras_namespace)s/%(extras_app_name)s]"
@@ -35,6 +36,11 @@ def filter(self, record):
         return 1
 
 
+class ExtraErrorFilter(logging.Filter):
+    def filter(self, record):
+        return record.levelno >= logging.ERROR
+
+
 class StatusFormatter(logging.Formatter):
     def __init__(self):
         super(StatusFormatter, self).__init__(_LOG_FORMAT, None)
@@ -74,6 +80,17 @@ def emit(self, record):
         append_log(record, self.format(record))
 
 
+class StatusErrorHandler(logging.Handler):
+    def __init__(self):
+        super(StatusErrorHandler, self).__init__(logging.ERROR)
+        self.addFilter(ExtraErrorFilter())
+        self.addFilter(ExtraFilter())
+        self.setFormatter(StatusFormatter())
+
+    def emit(self, record):
+        append_error_log(record, self.format(record))
+
+
 def set_extras(app_spec=None, app_name=None, namespace=None, deployment_id=None):
     if app_spec:
         app_name = app_spec.name
@@ -101,3 +118,19 @@ def append_log(record, message):
     if hasattr(record, "extras"):
         key = (record.extras.get("app_name"), record.extras.get("namespace"), record.extras.get("deployment_id"))
         _LOGS[key].append(message)
+
+
+def get_running_error_logs(app_name, namespace, deployment_id):
+    key = (app_name, namespace, deployment_id)
+    return _ERROR_LOGS.get(key, [])
+
+
+def get_final_error_logs(app_name, namespace, deployment_id):
+    key = (app_name, namespace, deployment_id)
+    return _ERROR_LOGS.pop(key, [])
+
+
+def append_error_log(record, message):
+    if hasattr(record, "extras"):
+        key = (record.extras.get("app_name"), record.extras.get("namespace"), record.extras.get("deployment_id"))
+        _ERROR_LOGS[key].append(message)
diff --git a/fiaas_deploy_daemon/logsetup.py b/fiaas_deploy_daemon/logsetup.py
index 355a5fef..37f89195 100644
--- a/fiaas_deploy_daemon/logsetup.py
+++ b/fiaas_deploy_daemon/logsetup.py
@@ -22,7 +22,7 @@
 import sys
 
 from fiaas_deploy_daemon.log_extras import StatusHandler
-from .log_extras import ExtraFilter
+from .log_extras import ExtraFilter, StatusErrorHandler
 
 
 class FiaasFormatter(logging.Formatter):
@@ -103,6 +103,7 @@ def init_logging(config):
         root.setLevel(logging.DEBUG)
     root.addHandler(_create_default_handler(config))
     root.addHandler(StatusHandler())
+    root.addHandler(StatusErrorHandler())
     _set_special_levels()
 
 
diff --git a/setup.py b/setup.py
index bf47f1f2..e8869d5c 100755
--- a/setup.py
+++ b/setup.py
@@ -33,9 +33,9 @@ def read(filename):
     "decorator < 5.0.0",  # 5.0.0 and later drops py2 support (transitive dep from pinject)
     "six >= 1.12.0",
     "dnspython == 1.16.0",
-    "k8s == 0.22.0",
+    "k8s == 0.23.4",
     "appdirs == 1.4.3",
-    "requests-toolbelt == 0.9.1",
+    "requests-toolbelt == 0.10.1",
     "backoff == 1.8.0",
     "py27hash == 1.1.0",
 ]
@@ -52,7 +52,8 @@ def read(filename):
 ]
 
 DEPLOY_REQ = [
-    "requests == 2.27.1",
+    "urllib3 == 1.26.17",
+    "requests == 2.31.0",
     "ipaddress == 1.0.22",  # Required by requests for resolving IP address in SSL cert
 ]
 
diff --git a/tests/fiaas_deploy_daemon/crd/test_crd_resources_syncer_apiextensionsv1.py b/tests/fiaas_deploy_daemon/crd/test_crd_resources_syncer_apiextensionsv1.py
index aab85aba..bbb11efd 100644
--- a/tests/fiaas_deploy_daemon/crd/test_crd_resources_syncer_apiextensionsv1.py
+++ b/tests/fiaas_deploy_daemon/crd/test_crd_resources_syncer_apiextensionsv1.py
@@ -46,6 +46,9 @@
                 "name": "v1",
                 "served": True,
                 "storage": True,
+                "subresources": {
+                    "status": {}
+                },
                 "schema": {
                     "openAPIV3Schema": {
                         "type": "object",
@@ -87,13 +90,29 @@
                                         },
                                     },
                                 },
+                            },
+                            "status": {
+                                "type": "object",
+                                "properties": {
+                                    "result": {
+                                        "type": "string"
+                                    },
+                                    "observedGeneration": {
+                                        "type": "integer"
+                                    },
+                                    "logs": {
+                                        "type": "array",
+                                        "items": {
+                                            "type": "string"
+                                        }
+                                    },
+                                    "deployment_id": {
+                                        "type": "string"
+                                    }
+                                }
                             }
                         },
-                        "oneOf": [],
-                        "allOf": [],
                         "required": [],
-                        "x-kubernetes-list-map-keys": [],
-                        "anyOf": [],
                     }
                 },
             }
diff --git a/tests/fiaas_deploy_daemon/crd/test_crd_status.py b/tests/fiaas_deploy_daemon/crd/test_crd_status.py
index 195315b1..cd685c97 100644
--- a/tests/fiaas_deploy_daemon/crd/test_crd_status.py
+++ b/tests/fiaas_deploy_daemon/crd/test_crd_status.py
@@ -25,7 +25,7 @@
 
 from fiaas_deploy_daemon.crd import status
 from fiaas_deploy_daemon.crd.status import _cleanup, OLD_STATUSES_TO_KEEP, LAST_UPDATED_KEY, now
-from fiaas_deploy_daemon.crd.types import FiaasApplicationStatus
+from fiaas_deploy_daemon.crd.types import FiaasApplicationStatus, FiaasApplication
 from fiaas_deploy_daemon.lifecycle import (
     DEPLOY_STATUS_CHANGED,
     STATUS_INITIATED,
@@ -50,6 +50,11 @@ def get_or_create(self):
         with mock.patch("fiaas_deploy_daemon.crd.status.FiaasApplicationStatus.get_or_create", spec_set=True) as m:
             yield m
 
+    @pytest.fixture
+    def get_app(self):
+        with mock.patch("fiaas_deploy_daemon.crd.status.FiaasApplication.get", spec_set=True) as m:
+            yield m
+
     @pytest.fixture
     def find(self):
         with mock.patch("fiaas_deploy_daemon.crd.status.FiaasApplicationStatus.find", spec_set=True) as m:
@@ -100,8 +105,9 @@ def pytest_generate_tests(self, metafunc):
                     test_id = "{} status on {}".format(action, result)
                     metafunc.addcall({"test_data": test_data}, test_id)
 
+    @pytest.mark.parametrize("same_deployment_id", (True, False))
     @pytest.mark.usefixtures("post", "put", "find", "logs")
-    def test_action_on_signal(self, request, get, app_spec, test_data, signal):
+    def test_action_on_signal(self, request, get, get_app, app_spec, test_data, signal, same_deployment_id):
         app_name = "{}-isb5oqum36ylo".format(test_data.result)
         expected_logs = [LOG_LINE]
         if not test_data.new:
@@ -128,13 +134,19 @@ def test_action_on_signal(self, request, get, app_spec, test_data, signal):
             }
             get.return_value = get_response
 
+        app_response = mock.create_autospec(FiaasApplication)
+        app_response.metadata.generation = 1
+        if same_deployment_id:
+            app_response.metadata.labels = {"fiaas/deployment_id": app_spec.deployment_id}
+        get_app.return_value = app_response
+
         # expected data used in expected api response and to configure mocks
         labels = app_spec.labels._replace(status={"status/label": "true"})
         annotations = app_spec.annotations._replace(status={"status/annotations": "true"})
         app_spec = app_spec._replace(name=test_data.result, labels=labels, annotations=annotations)
 
         # setup status signals
-        status.connect_signals()
+        status.connect_signals(True)
 
         # setup expected API call resulting from status update
         expected_call = {
@@ -192,6 +204,13 @@ def test_action_on_signal(self, request, get, app_spec, test_data, signal):
         called_mock.assert_called_once_with(url, expected_call)
         ignored_mock = request.getfixturevalue(test_data.ignored_mock)
         ignored_mock.assert_not_called()
+        if same_deployment_id:
+            assert app_response.status.result == test_data.result
+            assert app_response.status.observedGeneration == app_response.metadata.generation
+            assert app_response.status.deployment_id == app_spec.deployment_id
+            app_response.save_status.assert_called_once()
+        else:
+            app_response.save_status.assert_not_called()
 
     @pytest.mark.parametrize(
         "deployment_id",
@@ -231,7 +250,7 @@ def test_ignore_notfound_on_cleanup(self, find, delete, app_spec):
         ),
     )
     @pytest.mark.usefixtures("get", "post", "put", "find", "logs")
-    def test_retry_on_conflict(self, get_or_create, save, app_spec, signal, result, fail_times):
+    def test_retry_on_conflict(self, get_or_create, get_app, save, app_spec, signal, result, fail_times):
         def _fail():
             response = mock.MagicMock(spec=Response)
             response.status_code = 409  # Conflict
@@ -241,7 +260,12 @@ def _fail():
         application_status = FiaasApplicationStatus(metadata=ObjectMeta(name=app_spec.name, namespace="default"))
         get_or_create.return_value = application_status
 
-        status.connect_signals()
+        app_response = mock.create_autospec(FiaasApplication)
+        app_response.metadata.generation = 1
+        app_response.metadata.labels = {"fiaas/deployment_id": app_spec.deployment_id}
+        get_app.return_value = app_response
+
+        status.connect_signals(True)
         lifecycle_subject = _subject_from_app_spec(app_spec)
 
         try:
@@ -253,9 +277,11 @@ def _fail():
         save_calls = min(fail_times + 1, CONFLICT_MAX_RETRIES)
         assert save.call_args_list == [mock.call()] * save_calls
 
+        app_response.save_status.assert_called_once()
+
     @pytest.mark.parametrize("result", (STATUS_INITIATED, STATUS_STARTED, STATUS_SUCCESS, STATUS_FAILED))
     @pytest.mark.usefixtures("get", "post", "put", "find", "logs")
-    def test_fail_on_error(self, get_or_create, save, app_spec, signal, result):
+    def test_fail_on_save_error(self, get_or_create, get_app, save, app_spec, signal, result):
         response = mock.MagicMock(spec=Response)
         response.status_code = 403
 
@@ -264,7 +290,44 @@ def test_fail_on_error(self, get_or_create, save, app_spec, signal, result):
         application_status = FiaasApplicationStatus(metadata=ObjectMeta(name=app_spec.name, namespace="default"))
         get_or_create.return_value = application_status
 
-        status.connect_signals()
+        app_response = mock.create_autospec(FiaasApplication)
+        app_response.metadata.generation = 1
+        app_response.metadata.labels = {"fiaas/deployment_id": app_spec.deployment_id}
+        get_app.return_value = app_response
+
+        status.connect_signals(True)
+        lifecycle_subject = _subject_from_app_spec(app_spec)
+
+        with pytest.raises(ClientError):
+            signal(DEPLOY_STATUS_CHANGED).send(status=result, subject=lifecycle_subject)
+
+    @pytest.mark.parametrize("result", (STATUS_INITIATED, STATUS_STARTED, STATUS_SUCCESS, STATUS_FAILED))
+    @pytest.mark.usefixtures("get", "post", "put", "find", "logs")
+    def test_fail_on_save_status_error(self, get, get_app, save, app_spec, signal, result):
+        response = mock.MagicMock(spec=Response)
+        response.status_code = 403
+
+        app_response = mock.create_autospec(FiaasApplication)
+        app_response.metadata.generation = 1
+        app_response.metadata.labels = {"fiaas/deployment_id": app_spec.deployment_id}
+        app_response.save_status.side_effect = ClientError("No", response=response)
+        get_app.return_value = app_response
+
+        status.connect_signals(True)
+        lifecycle_subject = _subject_from_app_spec(app_spec)
+
+        with pytest.raises(ClientError):
+            signal(DEPLOY_STATUS_CHANGED).send(status=result, subject=lifecycle_subject)
+
+    @pytest.mark.parametrize("result", (STATUS_INITIATED, STATUS_STARTED, STATUS_SUCCESS, STATUS_FAILED))
+    @pytest.mark.usefixtures("get", "post", "put", "find", "logs")
+    def test_fail_on_get_app_error(self, get_or_create, get_app, save, app_spec, signal, result):
+        response = mock.MagicMock(spec=Response)
+        response.status_code = 403
+
+        get_app.side_effect = ClientError("No", response=response)
+
+        status.connect_signals(True)
         lifecycle_subject = _subject_from_app_spec(app_spec)
 
         with pytest.raises(ClientError):
diff --git a/tests/fiaas_deploy_daemon/crd/test_crd_watcher.py b/tests/fiaas_deploy_daemon/crd/test_crd_watcher.py
index fcf29355..0d109d52 100644
--- a/tests/fiaas_deploy_daemon/crd/test_crd_watcher.py
+++ b/tests/fiaas_deploy_daemon/crd/test_crd_watcher.py
@@ -39,6 +39,7 @@
             "name": "example",
             "namespace": "the-namespace",
             "uid": "c1f34517-6f54-11ea-8eaf-0ad3d9992c8c",
+            "generation": 1,
         },
         "spec": {
             "application": "example",
@@ -59,6 +60,29 @@
     "type": WatchEvent.DELETED,
 }
 
+STATUS_EVENT = {
+    "object": {
+        "metadata": {
+            "labels": {"fiaas/deployment_id": "deployment_id"},
+            "name": "example",
+            "namespace": "the-namespace",
+            "uid": "c1f34517-6f54-11ea-8eaf-0ad3d9992c8c",
+            "generation": 1,
+        },
+        "spec": {
+            "application": "example",
+            "config": {"version": 2, "host": "example.com", "namespace": "default", "annotations": {}},
+            "image": "example/app",
+        },
+        "status": {
+            "result": "SUCCEED",
+            "observedGeneration": 1,
+            "deployment_id": "deployment_id",
+        },
+    },
+    "type": WatchEvent.MODIFIED,
+}
+
 
 class FakeCrdResourcesSyncer(object):
     @classmethod
@@ -278,3 +302,10 @@ def test_deploy_based_on_status_result(self, crd_watcher, deploy_queue, watcher,
         assert deploy_queue.qsize() == 0
         crd_watcher._watch(None)
         assert deploy_queue.qsize() == count
+
+    def test_deploy_save_status(self, crd_watcher, deploy_queue, watcher, status_get):
+        watcher.watch.return_value = [WatchEvent(STATUS_EVENT, FiaasApplication)]
+
+        assert deploy_queue.qsize() == 0
+        crd_watcher._watch(None)
+        assert deploy_queue.qsize() == 0
diff --git a/tests/fiaas_deploy_daemon/test_bootstrap_e2e.py b/tests/fiaas_deploy_daemon/test_bootstrap_e2e.py
index bb40e235..d6345d9b 100644
--- a/tests/fiaas_deploy_daemon/test_bootstrap_e2e.py
+++ b/tests/fiaas_deploy_daemon/test_bootstrap_e2e.py
@@ -169,6 +169,7 @@ def run_bootstrap(self, request, kubernetes, k8s_version, use_docker_for_e2e):
         ]
         if use_apiextensionsv1_crd(k8s_version):
             args.append("--use-apiextensionsv1-crd")
+            args.append("--include-status-in-app")
         if use_networkingv1_ingress(k8s_version):
             args.append("--use-networkingv1-ingress")
 
diff --git a/tests/fiaas_deploy_daemon/test_config.py b/tests/fiaas_deploy_daemon/test_config.py
index d0aadcf7..e76fcc22 100644
--- a/tests/fiaas_deploy_daemon/test_config.py
+++ b/tests/fiaas_deploy_daemon/test_config.py
@@ -88,6 +88,7 @@ def test_infrastructure_param(self):
             "enable_deprecated_tls_entry_per_host",
             "disable_crd_creation",
             "datadog_activate_sleep",
+            "include_status_in_app",
         ),
     )
     def test_flags(self, key):
diff --git a/tests/fiaas_deploy_daemon/test_e2e.py b/tests/fiaas_deploy_daemon/test_e2e.py
index db1fba55..b3d8e0f5 100644
--- a/tests/fiaas_deploy_daemon/test_e2e.py
+++ b/tests/fiaas_deploy_daemon/test_e2e.py
@@ -182,12 +182,13 @@ def prepare_fdd(self, request, kubernetes, k8s_version, use_docker_for_e2e, serv
             "use-issuer.example.com=certmanager.k8s.io/issuer",
             "--use-ingress-tls",
             "default_off",
-            "--enable-crd-support",
+            "--enable-crd-support"
         ]
         if service_account:
             args.append("--enable-service-account-per-app")
         if use_apiextensionsv1_crd(k8s_version):
             args.append("--use-apiextensionsv1-crd")
+            args.append("--include-status-in-app")
         if use_networkingv1_ingress(k8s_version):
             args.append("--use-networkingv1-ingress")
         args = docker_args + args
@@ -492,6 +493,8 @@ def run_crd_deploy(self, custom_resource_definition, service_type, service_accou
         # Check that deployment status is RUNNING
         def _assert_status():
             status = FiaasApplicationStatus.get(create_name(name, DEPLOYMENT_ID1))
+            status_inline = FiaasApplication.get(name).status.result
+            assert status_inline == "RUNNING"
             assert status.result == "RUNNING"
             assert len(status.logs) > 0
             assert any("Saving result RUNNING for default/{}".format(name) in line for line in status.logs)
@@ -511,6 +514,8 @@ def _assert_status():
         )
 
         if not service_account:
+            # Get fiaas_application from server to avoid Conflict error
+            fiaas_application = FiaasApplication.get(name)
             # Redeploy, new image, possibly new init-container
             fiaas_application.spec.image = IMAGE2
             fiaas_application.metadata.labels["fiaas/deployment_id"] = DEPLOYMENT_ID2
@@ -616,6 +621,8 @@ def test_multiple_ingresses(self, request, input, expected, k8s_version):
         # Check that deployment status is RUNNING
         def _assert_status():
             status = FiaasApplicationStatus.get(create_name(name, DEPLOYMENT_ID1))
+            status_inline = FiaasApplication.get(name).status.result
+            assert status_inline == "RUNNING"
             assert status.result == "RUNNING"
             assert len(status.logs) > 0
             assert any("Saving result RUNNING for default/{}".format(name) in line for line in status.logs)
@@ -632,6 +639,8 @@ def _check_two_ingresses():
 
         wait_until(_check_two_ingresses, patience=PATIENCE)
 
+        # Get fiaas_application from server to avoid Conflict error
+        fiaas_application = FiaasApplication.get(name)
         # Remove 2nd ingress to make sure cleanup works
         fiaas_application.spec.config["ingress"].pop()
         if not fiaas_application.spec.config["ingress"]:
diff --git a/tests/fiaas_deploy_daemon/test_log_extras.py b/tests/fiaas_deploy_daemon/test_log_extras.py
index 47314973..142e4b17 100644
--- a/tests/fiaas_deploy_daemon/test_log_extras.py
+++ b/tests/fiaas_deploy_daemon/test_log_extras.py
@@ -18,9 +18,11 @@
 
 import pytest
 
-from fiaas_deploy_daemon.log_extras import StatusHandler, set_extras, get_final_logs
+from fiaas_deploy_daemon.log_extras import StatusHandler, set_extras, get_final_logs, StatusErrorHandler, \
+    get_final_error_logs
 
 TEST_MESSAGE = "This is a test log message"
+TEST_MESSAGE_ERROR = "This is a test log error message"
 
 
 class TestLogExtras(object):
@@ -38,6 +40,21 @@ def test_status_log_has_extra(self, app_spec):
         assert app_spec.name in log_message
         assert app_spec.namespace in log_message
 
+    def test_status_log_has_extra_and_error(self, app_spec):
+        set_extras(app_spec)
+        logger = logging.getLogger("test.log.extras")
+        logger.addHandler(StatusErrorHandler())
+        logger.warning(TEST_MESSAGE)
+        logger.error(TEST_MESSAGE_ERROR)
+        logs = get_final_error_logs(
+            app_name=app_spec.name, namespace=app_spec.namespace, deployment_id=app_spec.deployment_id
+        )
+        assert len(logs) == 1
+        log_message = logs[0]
+        assert TEST_MESSAGE_ERROR in log_message
+        assert app_spec.name in log_message
+        assert app_spec.namespace in log_message
+
     @pytest.fixture
     def spec(self, request, app_spec):
         if request.param: