diff --git a/content/testsuite/testsuite.go b/content/testsuite/testsuite.go
index c02498f4e..4371df208 100644
--- a/content/testsuite/testsuite.go
+++ b/content/testsuite/testsuite.go
@@ -73,10 +73,15 @@ func ContentCrossNSIsolatedSuite(t *testing.T, name string, storeFn StoreInitFn)
 	t.Run("CrossNamespaceIsolate", makeTest(t, name, storeFn, checkCrossNSIsolate))
 }
 
+// ContentSharedNSIsolatedSuite runs a test suite for shared namespaces under isolated content policy
+func ContentSharedNSIsolatedSuite(t *testing.T, name string, storeFn StoreInitFn) {
+	t.Run("SharedNamespaceIsolate", makeTest(t, name, storeFn, checkSharedNSIsolate))
+}
+
 // ContextWrapper is used to decorate new context used inside the test
 // before using the context on the content store.
 // This can be used to support leasing and multiple namespaces tests.
-type ContextWrapper func(ctx context.Context) (context.Context, func(context.Context) error, error)
+type ContextWrapper func(ctx context.Context, sharedNS bool) (context.Context, func(context.Context) error, error)
 
 type wrapperKey struct{}
 
@@ -121,7 +126,7 @@ func makeTest(t *testing.T, name string, storeFn func(ctx context.Context, root
 		w, ok := ctx.Value(wrapperKey{}).(ContextWrapper)
 		if ok {
 			var done func(context.Context) error
-			ctx, done, err = w(ctx)
+			ctx, done, err = w(ctx, false)
 			if err != nil {
 				t.Fatalf("Error wrapping context: %+v", err)
 			}
@@ -824,7 +829,7 @@ func checkCrossNSShare(ctx context.Context, t *testing.T, cs content.Store) {
 		t.Fatal(err)
 	}
 
-	ctx2, done, err := wrap(context.Background())
+	ctx2, done, err := wrap(context.Background(), false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -876,7 +881,7 @@ func checkCrossNSAppend(ctx context.Context, t *testing.T, cs content.Store) {
 		t.Fatal(err)
 	}
 
-	ctx2, done, err := wrap(context.Background())
+	ctx2, done, err := wrap(context.Background(), false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -945,7 +950,7 @@ func checkCrossNSIsolate(ctx context.Context, t *testing.T, cs content.Store) {
 	}
 	t2 := time.Now()
 
-	ctx2, done, err := wrap(context.Background())
+	ctx2, done, err := wrap(context.Background(), false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -962,6 +967,64 @@ func checkCrossNSIsolate(ctx context.Context, t *testing.T, cs content.Store) {
 	checkNewlyCreated(t, w, t1, t2, t3, t4)
 }
 
+func checkSharedNSIsolate(ctx context.Context, t *testing.T, cs content.Store) {
+	wrap, ok := ctx.Value(wrapperKey{}).(ContextWrapper)
+	if !ok {
+		t.Skip("multiple contexts not supported")
+	}
+
+	ctx1, done1, err := wrap(context.Background(), true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer done1(ctx1)
+
+	var size int64 = 1000
+	b, d := createContent(size)
+	ref := fmt.Sprintf("ref-%d", size)
+	t1 := time.Now()
+
+	if err := content.WriteBlob(ctx1, cs, ref, bytes.NewReader(b), ocispec.Descriptor{Size: size, Digest: d}); err != nil {
+		t.Fatal(err)
+	}
+
+	ctx2, done2, err := wrap(context.Background(), false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer done2(ctx2)
+
+	w, err := cs.Writer(ctx2, content.WithRef(ref), content.WithDescriptor(ocispec.Descriptor{Size: size, Digest: d}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer w.Close()
+	t2 := time.Now()
+
+	checkStatus(t, w, content.Status{
+		Ref:    ref,
+		Offset: size,
+		Total:  size,
+	}, d, t1, t2, t1, t2)
+
+	if err := w.Commit(ctx2, size, d); err != nil {
+		t.Fatal(err)
+	}
+	t3 := time.Now()
+
+	info := content.Info{
+		Digest: d,
+		Size:   size,
+	}
+	if err := checkContent(ctx1, cs, d, info, t1, t3, t1, t3); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := checkContent(ctx2, cs, d, info, t1, t3, t1, t3); err != nil {
+		t.Fatal(err)
+	}
+}
+
 func checkStatus(t *testing.T, w content.Writer, expected content.Status, d digest.Digest, preStart, postStart, preUpdate, postUpdate time.Time) {
 	t.Helper()
 	st, err := w.Status()
diff --git a/docs/ops.md b/docs/ops.md
index c884db083..66ead5fa2 100644
--- a/docs/ops.md
+++ b/docs/ops.md
@@ -232,3 +232,7 @@ The default is "shared". While this is largely the most desired policy, one can
 [plugins.bolt]
 	content_sharing_policy = "isolated"
 ```
+
+In "isolated" mode, it is also possible to share only the contents of a specific namespace by adding the label `containerd.io/namespace.shareable=true` to that namespace.
+This will make its blobs available in all other namespaces even if the content sharing policy is set to "isolated".
+If the label value is set to anything other than `true`, the namespace content will not be shared.
diff --git a/integration/client/content_test.go b/integration/client/content_test.go
index 6ad760e47..ffcb1b810 100644
--- a/integration/client/content_test.go
+++ b/integration/client/content_test.go
@@ -41,7 +41,7 @@ func newContentStore(ctx context.Context, root string) (context.Context, content
 		name  = testsuite.Name(ctx)
 	)
 
-	wrap := func(ctx context.Context) (context.Context, func(context.Context) error, error) {
+	wrap := func(ctx context.Context, sharedNS bool) (context.Context, func(context.Context) error, error) {
 		n := atomic.AddUint64(&count, 1)
 		ctx = namespaces.WithNamespace(ctx, fmt.Sprintf("%s-n%d", name, n))
 		return client.WithLease(ctx)
diff --git a/labels/labels.go b/labels/labels.go
index d76ff2cf9..6be006c3e 100644
--- a/labels/labels.go
+++ b/labels/labels.go
@@ -19,3 +19,7 @@ package labels
 // LabelUncompressed is added to compressed layer contents.
 // The value is digest of the uncompressed content.
 const LabelUncompressed = "containerd.io/uncompressed"
+
+// LabelSharedNamespace is added to a namespace to allow that namespaces
+// contents to be shared.
+const LabelSharedNamespace = "containerd.io/namespace.shareable"
diff --git a/metadata/content.go b/metadata/content.go
index 66d0ee263..2df665fcf 100644
--- a/metadata/content.go
+++ b/metadata/content.go
@@ -398,7 +398,7 @@ func (cs *contentStore) Writer(ctx context.Context, opts ...content.WriterOpt) (
 				return nil
 			}
 
-			if cs.shared {
+			if cs.shared || isSharedContent(tx, wOpts.Desc.Digest) {
 				if st, err := cs.Store.Info(ctx, wOpts.Desc.Digest); err == nil {
 					// Ensure the expected size is the same, it is likely
 					// an error if the size is mismatched but the caller
@@ -706,6 +706,33 @@ func (cs *contentStore) checkAccess(ctx context.Context, dgst digest.Digest) err
 	})
 }
 
+func isSharedContent(tx *bolt.Tx, dgst digest.Digest) bool {
+	v1bkt := tx.Bucket(bucketKeyVersion)
+	if v1bkt == nil {
+		return false
+	}
+	// iterate through each namespace
+	v1c := v1bkt.Cursor()
+	for nk, _ := v1c.First(); nk != nil; nk, _ = v1c.Next() {
+		ns := string(nk)
+		lbkt := getNamespaceLabelsBucket(tx, ns)
+		if lbkt == nil {
+			continue
+		}
+		// iterate through each label
+		lbc := lbkt.Cursor()
+		for k, v := lbc.First(); k != nil; k, v = lbc.Next() {
+			if string(k) == labels.LabelSharedNamespace {
+				if string(v) == "true" && getBlobBucket(tx, ns, dgst) != nil {
+					return true
+				}
+				break
+			}
+		}
+	}
+	return false
+}
+
 func validateInfo(info *content.Info) error {
 	for k, v := range info.Labels {
 		if err := labels.Validate(k, v); err != nil {
diff --git a/metadata/content_test.go b/metadata/content_test.go
index 061095c0f..8e716aebd 100644
--- a/metadata/content_test.go
+++ b/metadata/content_test.go
@@ -29,6 +29,7 @@ import (
 	"github.com/containerd/containerd/content/local"
 	"github.com/containerd/containerd/content/testsuite"
 	"github.com/containerd/containerd/errdefs"
+	"github.com/containerd/containerd/labels"
 	"github.com/containerd/containerd/leases"
 	"github.com/containerd/containerd/namespaces"
 	digest "github.com/opencontainers/go-digest"
@@ -52,9 +53,18 @@ func createContentStore(ctx context.Context, root string, opts ...DBOpt) (contex
 		count uint64
 		name  = testsuite.Name(ctx)
 	)
-	wrap := func(ctx context.Context) (context.Context, func(context.Context) error, error) {
+	wrap := func(ctx context.Context, sharedNS bool) (context.Context, func(context.Context) error, error) {
 		n := atomic.AddUint64(&count, 1)
-		return namespaces.WithNamespace(ctx, fmt.Sprintf("%s-n%d", name, n)), func(context.Context) error {
+		ctx2 := namespaces.WithNamespace(ctx, fmt.Sprintf("%s-n%d", name, n))
+		if sharedNS {
+			db.Update(func(tx *bolt.Tx) error {
+				if ns, err := namespaces.NamespaceRequired(ctx2); err == nil {
+					return NewNamespaceStore(tx).SetLabel(ctx2, ns, labels.LabelSharedNamespace, "true")
+				}
+				return err
+			})
+		}
+		return ctx2, func(context.Context) error {
 			return nil
 		}, nil
 	}
@@ -78,6 +88,10 @@ func TestContent(t *testing.T) {
 		t, "metadata", createContentStoreWithPolicy([]DBOpt{
 			WithPolicyIsolated,
 		}...))
+	testsuite.ContentSharedNSIsolatedSuite(
+		t, "metadata", createContentStoreWithPolicy([]DBOpt{
+			WithPolicyIsolated,
+		}...))
 }
 
 func TestContentLeased(t *testing.T) {