diff --git a/kubernetes-deploy.gemspec b/kubernetes-deploy.gemspec
index c01dbd3bd..968f72d6b 100644
--- a/kubernetes-deploy.gemspec
+++ b/kubernetes-deploy.gemspec
@@ -24,12 +24,13 @@ Gem::Specification.new do |spec|
 
   spec.required_ruby_version = '>= 2.3.0'
   spec.add_dependency "activesupport", ">= 4.2"
-  spec.add_dependency "kubeclient", "~> 2.4"
+  spec.add_dependency "kubeclient", "~> 2.5.1"
   spec.add_dependency "rest-client", ">= 1.7" # Minimum required by kubeclient. Remove when kubeclient releases v3.0.
   spec.add_dependency "googleauth", ">= 0.5"
   spec.add_dependency "ejson", "1.0.1"
   spec.add_dependency "colorize", "~> 0.8"
   spec.add_dependency "statsd-instrument", "~> 2.1"
+  spec.add_dependency "jsonpath", "0.8.8"
 
   spec.add_development_dependency "bundler"
   spec.add_development_dependency "rake", "~> 10.0"
diff --git a/lib/kubernetes-deploy/deploy_task.rb b/lib/kubernetes-deploy/deploy_task.rb
index 88bbef6fc..bc5633787 100644
--- a/lib/kubernetes-deploy/deploy_task.rb
+++ b/lib/kubernetes-deploy/deploy_task.rb
@@ -5,8 +5,11 @@
 require 'yaml'
 require 'shellwords'
 require 'tempfile'
+require 'kubernetes-deploy/discoverable_resource'
 require 'kubernetes-deploy/kubernetes_resource'
+
 %w(
+  genericresource
   cloudsql
   config_map
   deployment
@@ -28,6 +31,8 @@
   topic
   bucket
   stateful_set
+  customresourcedefinition
+  thirdpartyresource
 ).each do |subresource|
   require "kubernetes-deploy/kubernetes_resource/#{subresource}"
 end
@@ -77,6 +82,18 @@ class DeployTask
       autoscaling/v1/HorizontalPodAutoscaler
     ).freeze
 
+    def prune_whitelist
+      resources = DiscoverableResource.all.select(&:prunable?)
+      identities = resources.map(&:identity)
+      PRUNE_WHITELIST + identities
+    end
+
+    def predeploy_sequence
+      resources = DiscoverableResource.all.select(&:predeploy?)
+      identities = resources.map(&:identity)
+      PREDEPLOY_SEQUENCE + identities
+    end
+
     NOT_FOUND_ERROR = 'NotFound'
 
     def initialize(namespace:, context:, current_sha:, template_dir:, logger:, kubectl_instance: nil, bindings: {})
@@ -174,11 +191,11 @@ def find_bad_files_from_kubectl_output(stderr)
     end
 
     def deploy_has_priority_resources?(resources)
-      resources.any? { |r| PREDEPLOY_SEQUENCE.include?(r.type) }
+      resources.any? { |r| predeploy_sequence.include?(r.type) }
     end
 
     def predeploy_priority_resources(resource_list)
-      PREDEPLOY_SEQUENCE.each do |resource_type|
+      predeploy_sequence.each do |resource_type|
         matching_resources = resource_list.select { |r| r.type == resource_type }
         next if matching_resources.empty?
         deploy_resources(matching_resources, verify: true, record_summary: false)
@@ -206,13 +223,15 @@ def validate_definitions(resources)
 
     def discover_resources
       resources = []
+      # Explicitly discovering will discard all cached resources.
+      DiscoverableResource.discover(context: @context, logger: @logger)
       @logger.info("Discovering templates:")
 
       Dir.foreach(@template_dir) do |filename|
         next unless filename.end_with?(".yml.erb", ".yml", ".yaml", ".yaml.erb")
 
         split_templates(filename) do |r_def|
-          r = KubernetesResource.build(namespace: @namespace, context: @context, logger: @logger, definition: r_def)
+          r = DiscoverableResource.build(namespace: @namespace, context: @context, logger: @logger, definition: r_def)
           resources << r
           @logger.info "  - #{r.id}"
         end
@@ -376,7 +395,7 @@ def apply_all(resources, prune)
 
       if prune
         command.push("--prune", "--all")
-        PRUNE_WHITELIST.each { |type| command.push("--prune-whitelist=#{type}") }
+        prune_whitelist.each { |type| command.push("--prune-whitelist=#{type}") }
       end
 
       out, err, st = kubectl.run(*command, log_failure: false)
diff --git a/lib/kubernetes-deploy/discoverable_resource.rb b/lib/kubernetes-deploy/discoverable_resource.rb
new file mode 100644
index 000000000..80ce0f4bb
--- /dev/null
+++ b/lib/kubernetes-deploy/discoverable_resource.rb
@@ -0,0 +1,171 @@
+# frozen_string_literal: true
+require 'kubernetes-deploy/kubernetes_resource'
+require 'kubernetes-deploy/kubeclient_builder'
+require "pry"
+require 'jsonpath'
+
+module KubernetesDeploy
+  class DiscoverableResource < KubernetesResource
+    extend KubernetesDeploy::KubeclientBuilder
+
+    class << self
+      attr_accessor :version, :group, :type
+      attr_reader :timeout
+    end
+
+    STATUS_FIELD_ANNOTATION = 'kubernetes-deploy.shopify.io/status-field'
+    STATUS_SUCCESS_ANNOTATION = 'kubernetes-deploy.shopify.io/status-success'
+    TIMEOUT_ANNOTATION = 'kubernetes-deploy.shopify.io/timeout'
+    PREDEPLOY_ANNOTATION = 'kubernetes-deploy.shopify.io/predeploy'
+    PRUNABLE_ANNOTATION = 'kubernetes-deploy.shopify.io/prunable'
+
+    def type
+      self.class.type
+    end
+
+    def self.inherited(child_class)
+      child_classes.add child_class
+    end
+
+    def self.child_classes
+      @child_classes ||= Set.new
+    end
+
+    def self.prunable?
+      @prunable
+    end
+
+    def self.predeploy?
+      @predeploy
+    end
+
+    def self.identity
+      "#{group}/#{version}/#{type}"
+    end
+
+    def self.all
+      child_classes.dup
+    end
+
+    def self.discover(context:, logger:)
+      logger.info("Discovering custom resources:")
+      @resources = nil
+      @child_classes = nil
+      discover_tpr(v1beta1_kubeclient(context))
+      begin
+        discover_crd(v1beta1_crd_kubeclient(context))
+      rescue KubeException => err
+        logger.warn("Unable to discover CustomResourceDefinitions: #{err}")
+      end
+    end
+
+    def self.build(namespace:, context:, definition:, logger:)
+      return super if KubernetesDeploy.const_defined?(definition["kind"])
+
+      # We only discover once per kubernetes-deploy invocation
+      discover(context: context, logger: logger) unless @resources
+
+      type = definition["kind"]
+      group, _, version = definition['apiVersion'].rpartition('/')
+
+      resource_class = @resources.dig(group, type, version) if @resources
+      return super unless resource_class
+
+      opts = { namespace: namespace, context: context, definition: definition, logger: logger }
+      resource_class.new(**opts)
+    end
+
+    def self.discover_tpr(client)
+      return unless client.respond_to? :get_third_party_resources
+      resources = client.get_third_party_resources
+      resources.each do |res|
+        type, _, group = res.metadata.name.partition('.')
+        # TPR API supports multiple versions for a single resource :(
+        res.versions.each do |version|
+          discovered(group: group,
+                     type: type,
+                     version: version.name,
+                     annotations: res.metadata.annotations)
+        end
+      end
+    end
+
+    def self.discover_crd(client)
+      return unless client.respond_to? :get_custom_resource_definitions
+      resources = client.get_custom_resource_definitions
+      resources.each do |res|
+        discovered(group: res.spec.group,
+                   type: res.spec.names.kind,
+                   version: res.spec.version,
+                   annotations: res.metadata.annotations)
+      end
+    end
+
+    def self.discovered(group:, type:, version:, annotations:)
+      resource_class = Class.new(self) do
+        type = type.capitalize # TPRs are inconsistent about capitalization
+        @group = group
+        @type = type
+        @version = version
+        @prunable = DiscoverableResource.parse_bool(annotations[PRUNABLE_ANNOTATION])
+        @predeploy = DiscoverableResource.parse_bool(annotations[PREDEPLOY_ANNOTATION])
+        @timeout = DiscoverableResource.parse_timeout(type, annotations[TIMEOUT_ANNOTATION])
+
+        status_field = annotations[STATUS_FIELD_ANNOTATION]
+        success_status = annotations[STATUS_SUCCESS_ANNOTATION]
+
+        define_method 'deploy_succeeded?' do
+          getter = "get_#{type.downcase}"
+          @client ||= DiscoverableResource.kubeclient(context: @context, resource_class: self.class)
+          raw_json = @client.send(getter, @name, @namespace, as: :raw)
+          query_path = JsonPath.new(status_field)
+          current_status = query_path.first(raw_json)
+          current_status == success_status
+        end if status_field && success_status
+      end
+
+      add_resource(resource_class)
+    end
+
+    def self.add_resource(resource_class)
+      group = resource_class.group
+      type = resource_class.type
+      version = resource_class.version
+      @resources ||= {}
+      @resources[group] ||= {}
+      @resources[group][type] ||= {}
+      @resources[group][type][version] = resource_class
+    end
+
+    def self.parse_bool(value)
+      value.to_s == "true" # value could be nil
+    end
+
+    def self.parse_timeout(type, timeout)
+      ActiveSupport::Duration.parse(timeout) if timeout
+    rescue ActiveSupport::Duration::ISO8601Parser::ParsingError
+      raise FatalDeploymentError,
+        "Resource #{type} specified invalid timeout value '#{timeout}', must use ISO8601 duration."
+    end
+
+    def self.kubeclient(context:, resource_class:)
+      _build_kubeclient(
+        api_version: resource_class.version,
+        context: context,
+        endpoint_path: "/apis/#{resource_class.group}"
+      )
+    end
+
+    def self.v1beta1_kubeclient(context)
+      @v1beta1_kubeclient ||= build_v1beta1_kubeclient(context)
+    end
+
+    def self.v1beta1_crd_kubeclient(context)
+      @v1beta1_kubeclient_crd ||= _build_kubeclient(
+        api_version: "v1beta1",
+        context: context,
+        endpoint_path: "/apis/apiextensions.k8s.io/"
+      )
+    end
+  end
+end
diff --git a/lib/kubernetes-deploy/kubernetes_resource/config_map.rb b/lib/kubernetes-deploy/kubernetes_resource/config_map.rb
index 861373b55..867fbad98 100644
--- a/lib/kubernetes-deploy/kubernetes_resource/config_map.rb
+++ b/lib/kubernetes-deploy/kubernetes_resource/config_map.rb
@@ -1,28 +1,6 @@
 # frozen_string_literal: true
 module KubernetesDeploy
-  class ConfigMap < KubernetesResource
+  class ConfigMap < GenericResource
     TIMEOUT = 30.seconds
-
-    def sync
-      _, _err, st = kubectl.run("get", type, @name)
-      @status = st.success? ? "Available" : "Unknown"
-      @found = st.success?
-    end
-
-    def deploy_succeeded?
-      exists?
-    end
-
-    def deploy_failed?
-      false
-    end
-
-    def timeout_message
-      UNUSUAL_FAILURE_MESSAGE
-    end
-
-    def exists?
-      @found
-    end
   end
 end
diff --git a/lib/kubernetes-deploy/kubernetes_resource/customresourcedefinition.rb b/lib/kubernetes-deploy/kubernetes_resource/customresourcedefinition.rb
new file mode 100644
index 000000000..2cb8c25cc
--- /dev/null
+++ b/lib/kubernetes-deploy/kubernetes_resource/customresourcedefinition.rb
@@ -0,0 +1,6 @@
+# frozen_string_literal: true
+module KubernetesDeploy
+  class CustomResourceDefinition < GenericResource
+    TIMEOUT = 10.seconds
+  end
+end
diff --git a/lib/kubernetes-deploy/kubernetes_resource/genericresource.rb b/lib/kubernetes-deploy/kubernetes_resource/genericresource.rb
new file mode 100644
index 000000000..a9ebebc55
--- /dev/null
+++ b/lib/kubernetes-deploy/kubernetes_resource/genericresource.rb
@@ -0,0 +1,26 @@
+# frozen_string_literal: true
+module KubernetesDeploy
+  class GenericResource < KubernetesResource
+    def sync
+      _, _err, st = kubectl.run("get", type, @name)
+      @status = st.success? ? "Available" : "Unknown"
+      @found = st.success?
+    end
+
+    def deploy_succeeded?
+      exists?
+    end
+
+    def deploy_failed?
+      false
+    end
+
+    def timeout_message
+      UNUSUAL_FAILURE_MESSAGE
+    end
+
+    def exists?
+      @found
+    end
+  end
+end
diff --git a/lib/kubernetes-deploy/kubernetes_resource/ingress.rb b/lib/kubernetes-deploy/kubernetes_resource/ingress.rb
index 790284873..f1b2ecc60 100644
--- a/lib/kubernetes-deploy/kubernetes_resource/ingress.rb
+++ b/lib/kubernetes-deploy/kubernetes_resource/ingress.rb
@@ -1,24 +1,6 @@
 # frozen_string_literal: true
 module KubernetesDeploy
-  class Ingress < KubernetesResource
+  class Ingress < GenericResource
     TIMEOUT = 30.seconds
-
-    def sync
-      _, _err, st = kubectl.run("get", type, @name)
-      @status = st.success? ? "Created" : "Unknown"
-      @found = st.success?
-    end
-
-    def deploy_succeeded?
-      exists?
-    end
-
-    def deploy_failed?
-      false
-    end
-
-    def exists?
-      @found
-    end
   end
 end
diff --git a/lib/kubernetes-deploy/kubernetes_resource/pod_disruption_budget.rb b/lib/kubernetes-deploy/kubernetes_resource/pod_disruption_budget.rb
index 6753b9976..cb448dd92 100644
--- a/lib/kubernetes-deploy/kubernetes_resource/pod_disruption_budget.rb
+++ b/lib/kubernetes-deploy/kubernetes_resource/pod_disruption_budget.rb
@@ -1,29 +1,9 @@
 # frozen_string_literal: true
 module KubernetesDeploy
-  class PodDisruptionBudget < KubernetesResource
-    TIMEOUT = 10.seconds
-
-    def sync
-      _, _err, st = kubectl.run("get", type, @name)
-      @found = st.success?
-      @status = @found ? "Available" : "Unknown"
-    end
-
-    def deploy_succeeded?
-      exists?
-    end
-
+  class PodDisruptionBudget < GenericResource
     def deploy_method
       # Required until https://github.com/kubernetes/kubernetes/issues/45398 changes
       :replace_force
     end
-
-    def timeout_message
-      UNUSUAL_FAILURE_MESSAGE
-    end
-
-    def exists?
-      @found
-    end
   end
 end
diff --git a/lib/kubernetes-deploy/kubernetes_resource/pod_template.rb b/lib/kubernetes-deploy/kubernetes_resource/pod_template.rb
index 673ab8711..3a7107c2f 100644
--- a/lib/kubernetes-deploy/kubernetes_resource/pod_template.rb
+++ b/lib/kubernetes-deploy/kubernetes_resource/pod_template.rb
@@ -1,26 +1,5 @@
 # frozen_string_literal: true
 module KubernetesDeploy
-  class PodTemplate < KubernetesResource
-    def sync
-      _, _err, st = kubectl.run("get", type, @name)
-      @status = st.success? ? "Available" : "Unknown"
-      @found = st.success?
-    end
-
-    def deploy_succeeded?
-      exists?
-    end
-
-    def deploy_failed?
-      false
-    end
-
-    def timeout_message
-      UNUSUAL_FAILURE_MESSAGE
-    end
-
-    def exists?
-      @found
-    end
+  class PodTemplate < GenericResource
   end
 end
diff --git a/lib/kubernetes-deploy/kubernetes_resource/resource_quota.rb b/lib/kubernetes-deploy/kubernetes_resource/resource_quota.rb
index b98fbb310..cea2f5f7c 100644
--- a/lib/kubernetes-deploy/kubernetes_resource/resource_quota.rb
+++ b/lib/kubernetes-deploy/kubernetes_resource/resource_quota.rb
@@ -1,6 +1,6 @@
 # frozen_string_literal: true
 module KubernetesDeploy
-  class ResourceQuota < KubernetesResource
+  class ResourceQuota < GenericResource
     TIMEOUT = 30.seconds
 
     def sync
@@ -17,17 +17,5 @@ def sync
     def deploy_succeeded?
       @rollout_data.dig("spec", "hard") == @rollout_data.dig("status", "hard")
     end
-
-    def deploy_failed?
-      false
-    end
-
-    def timeout_message
-      UNUSUAL_FAILURE_MESSAGE
-    end
-
-    def exists?
-      @found
-    end
   end
 end
diff --git a/lib/kubernetes-deploy/kubernetes_resource/thirdpartyresource.rb b/lib/kubernetes-deploy/kubernetes_resource/thirdpartyresource.rb
new file mode 100644
index 000000000..073d4d58e
--- /dev/null
+++ b/lib/kubernetes-deploy/kubernetes_resource/thirdpartyresource.rb
@@ -0,0 +1,12 @@
+# frozen_string_literal: true
+module KubernetesDeploy
+  class ThirdPartyResource < GenericResource
+    TIMEOUT = 30.seconds
+
+    def exists?
+      # TPRs take time to become available.
+      _, _err, st = kubectl.run("get", @name)
+      st.success?
+    end
+  end
+end
diff --git a/test/fixtures/resource-discovery/definitions/crd.yml b/test/fixtures/resource-discovery/definitions/crd.yml
new file mode 100644
index 000000000..7ed28bbcb
--- /dev/null
+++ b/test/fixtures/resource-discovery/definitions/crd.yml
@@ -0,0 +1,18 @@
+---
+apiVersion: apiextensions.k8s.io/v1beta1
+kind: CustomResourceDefinition
+metadata:
+  name: widgets.api.foobar.com
+  annotations:
+    kubernetes-deploy.shopify.io/prunable: "true"
+    kubernetes-deploy.shopify.io/predeploy: "true"
+    kubernetes-deploy.shopify.io/timeout: "PT9M"
+    kubernetes-deploy.shopify.io/status-field: "status"
+    kubernetes-deploy.shopify.io/status-success: "ok"
+spec:
+  group: api.foobar.com
+  version: v1
+  names:
+    kind: Widget
+    plural: widgets
+  scope: Namespaced
diff --git a/test/fixtures/resource-discovery/definitions/crd_invalid_timespec.yml b/test/fixtures/resource-discovery/definitions/crd_invalid_timespec.yml
new file mode 100644
index 000000000..2d511f9cd
--- /dev/null
+++ b/test/fixtures/resource-discovery/definitions/crd_invalid_timespec.yml
@@ -0,0 +1,18 @@
+---
+apiVersion: apiextensions.k8s.io/v1beta1
+kind: CustomResourceDefinition
+metadata:
+  name: widgets.api.foobar.com
+  annotations:
+    kubernetes-deploy.shopify.io/prunable: "true"
+    kubernetes-deploy.shopify.io/predeploy: "true"
+    kubernetes-deploy.shopify.io/timeout: "foobar"
+    kubernetes-deploy.shopify.io/status-field: "status"
+    kubernetes-deploy.shopify.io/status-success: "ok"
+spec:
+  group: api.foobar.com
+  version: v1
+  names:
+    kind: Widget
+    plural: widgets
+  scope: Namespaced
diff --git a/test/fixtures/resource-discovery/definitions/crd_non_prunable_no_predeploy.yml b/test/fixtures/resource-discovery/definitions/crd_non_prunable_no_predeploy.yml
new file mode 100644
index 000000000..b1f04d0d1
--- /dev/null
+++ b/test/fixtures/resource-discovery/definitions/crd_non_prunable_no_predeploy.yml
@@ -0,0 +1,15 @@
+---
+apiVersion: apiextensions.k8s.io/v1beta1
+kind: CustomResourceDefinition
+metadata:
+  name: widgets.api.foobar.com
+  annotations:
+    kubernetes-deploy.shopify.io/status-field: "status"
+    kubernetes-deploy.shopify.io/status-success: "ok"
+spec:
+  group: api.foobar.com
+  version: v1
+  names:
+    kind: Widget
+    plural: widgets
+  scope: Namespaced
diff --git a/test/fixtures/resource-discovery/definitions/tpr.yml b/test/fixtures/resource-discovery/definitions/tpr.yml
new file mode 100644
index 000000000..92fbcf6f9
--- /dev/null
+++ b/test/fixtures/resource-discovery/definitions/tpr.yml
@@ -0,0 +1,17 @@
+---
+apiVersion: extensions/v1beta1
+description: Implements a dummy prunable resource
+kind: ThirdPartyResource
+metadata:
+  name: gizmo.api.prunable.com
+  annotations:
+    kubernetes-deploy.shopify.io/prunable: "true"
+    kubernetes-deploy.shopify.io/predeploy: "true"
+    kubernetes-deploy.shopify.io/timeout: "PT9M"
+    kubernetes-deploy.shopify.io/status-field: "status"
+    kubernetes-deploy.shopify.io/status-success: "ok"
+  labels:
+    k8s-deploy/prunable: "true"
+versions:
+- name: v1
+- name: v2
diff --git a/test/fixtures/resource-discovery/instances/crd.yml b/test/fixtures/resource-discovery/instances/crd.yml
new file mode 100644
index 000000000..e08049694
--- /dev/null
+++ b/test/fixtures/resource-discovery/instances/crd.yml
@@ -0,0 +1,6 @@
+---
+apiVersion: "api.foobar.com/v1"
+kind: Widget
+metadata:
+  name: my-first-widget
+status: "ok"
diff --git a/test/fixtures/resource-discovery/instances/tpr.yml b/test/fixtures/resource-discovery/instances/tpr.yml
new file mode 100644
index 000000000..27708032b
--- /dev/null
+++ b/test/fixtures/resource-discovery/instances/tpr.yml
@@ -0,0 +1,6 @@
+---
+apiVersion: "api.prunable.com/v1"
+kind: "Gizmo"
+metadata:
+  name: "my-first-gizmo"
+status: "ok"
diff --git a/test/integration/resource_discovery_test.rb b/test/integration/resource_discovery_test.rb
new file mode 100644
index 000000000..17975ed05
--- /dev/null
+++ b/test/integration/resource_discovery_test.rb
@@ -0,0 +1,98 @@
+# frozen_string_literal: true
+require 'test_helper'
+
+class ResourceDiscoveryTest < KubernetesDeploy::IntegrationTest
+  def kubectl
+    @kubectl ||= KubernetesDeploy::Kubectl.new(namespace: @namespace,
+      context: KubeclientHelper::MINIKUBE_CONTEXT,
+      logger:
+        KubernetesDeploy::FormattedLogger.build(
+          @namespace,
+          KubeclientHelper::MINIKUBE_CONTEXT,
+          $stdout
+        ),
+      log_failure_by_default: false)
+  end
+
+  def cleanup(*resources)
+    resources.each do |res|
+      _, err, st = kubectl.run("delete", res, "--all")
+      flunk(err) unless st.success?
+    end
+  end
+
+  def has_resource?(res)
+    _, _, st = kubectl.run("get", res)
+    st.success?
+  end
+
+  def has_tpr_support?
+    has_resource? "thirdpartyresources"
+  end
+
+  def has_crd_support?
+    has_resource? "customresourcedefinitions"
+  end
+
+  def test_prunable_tpr
+    skip unless has_tpr_support?
+    begin
+      assert_deploy_success(deploy_fixtures("resource-discovery/definitions", subset: ["tpr.yml"]))
+      assert_deploy_success(deploy_fixtures("resource-discovery/instances", subset: ["tpr.yml"]))
+      # Deploy any other resource to trigger pruning
+      assert_deploy_success(deploy_fixtures("hello-cloud", subset: ["configmap-data.yml",]))
+
+      assert_logs_match("The following resources were pruned: gizmo \"my-first-gizmo\"")
+      refute_logs_match("Don't know how to monitor resources of type Gizmo. " \
+                        "Assuming Gizmo/my-first-gizmo deployed successfully")
+    ensure
+      cleanup('thirdpartyresources')
+    end
+  end
+
+  def test_non_prunable_crd_no_predeploy
+    skip unless has_crd_support?
+    begin
+      assert_deploy_success(deploy_fixtures("resource-discovery/definitions",
+        subset: ["crd_non_prunable_no_predeploy.yml"]))
+      assert_deploy_success(deploy_fixtures("resource-discovery/instances", subset: ["crd.yml"]))
+      # Deploy any other non-priority (predeployable) resource to trigger pruning
+      assert_deploy_success(deploy_fixtures("hello-cloud", subset: ["daemon_set.yml",]))
+
+      refute_logs_match("The following resources were pruned: widget \"my-first-widget\"")
+      refute_logs_match("Don't know how to monitor resources of type Widget. " \
+                        "Assuming Widget/my-first-widget deployed successfully")
+      refute_logs_match("Predeploying priority resources")
+    ensure
+      cleanup('customresourcedefinitions')
+    end
+  end
+
+  def test_prunable_crd
+    skip unless has_crd_support?
+    begin
+      assert_deploy_success(deploy_fixtures("resource-discovery/definitions", subset: ["crd.yml"]))
+      assert_deploy_success(deploy_fixtures("resource-discovery/instances", subset: ["crd.yml"]))
+      # Deploy any other resource to trigger pruning
+      assert_deploy_success(deploy_fixtures("hello-cloud", subset: ["configmap-data.yml",]))
+
+      assert_logs_match("The following resources were pruned: widget \"my-first-widget\"")
+      refute_logs_match("Don't know how to monitor resources of type Widget. " \
+                        "Assuming Widget/my-first-widget deployed successfully")
+    ensure
+      cleanup('customresourcedefinitions')
+    end
+  end
+
+  def test_invalid_timeout_format
+    skip unless has_crd_support?
+    begin
+      assert_deploy_success(deploy_fixtures("resource-discovery/definitions", subset: ["crd_invalid_timespec.yml"]))
+      assert_deploy_failure(deploy_fixtures("resource-discovery/instances", subset: ["crd.yml"]))
+
+      assert_logs_match("Resource widget specified invalid timeout value 'foobar', must use iso8601 duration.")
+    ensure
+      cleanup('customresourcedefinitions')
+    end
+  end
+end