diff --git a/cli/command_snapshot_create.go b/cli/command_snapshot_create.go
index 3fd4daebd..48d224363 100644
--- a/cli/command_snapshot_create.go
+++ b/cli/command_snapshot_create.go
@@ -38,6 +38,7 @@ var (
 	snapshotCreateForceEnableActions      = snapshotCreateCommand.Flag("force-enable-actions", "Enable snapshot actions even if globally disabled on this client").Hidden().Bool()
 	snapshotCreateForceDisableActions     = snapshotCreateCommand.Flag("force-disable-actions", "Disable snapshot actions even if globally enabled on this client").Hidden().Bool()
 	snapshotCreateStdinFileName           = snapshotCreateCommand.Flag("stdin-file", "File path to be used for stdin data snapshot.").String()
+	snapshotCreateTags                    = snapshotCreateCommand.Flag("tags", "Tags applied on the snapshot. Must be provided in the <key>:<value> format.").Strings()
 )
 
 func runSnapshotCommand(ctx context.Context, rep repo.RepositoryWriter) error {
@@ -72,6 +73,11 @@ func runSnapshotCommand(ctx context.Context, rep repo.RepositoryWriter) error {
 
 	var finalErrors []string
 
+	tags, err := getTags(*snapshotCreateTags)
+	if err != nil {
+		return err
+	}
+
 	for _, snapshotDir := range sources {
 		if u.IsCanceled() {
 			log(ctx).Infof("Upload canceled")
@@ -89,7 +95,7 @@ func runSnapshotCommand(ctx context.Context, rep repo.RepositoryWriter) error {
 			UserName: rep.ClientOptions().Username,
 		}
 
-		if err := snapshotSingleSource(ctx, rep, u, sourceInfo); err != nil {
+		if err := snapshotSingleSource(ctx, rep, u, sourceInfo, tags); err != nil {
 			finalErrors = append(finalErrors, err.Error())
 		}
 	}
@@ -105,6 +111,30 @@ func runSnapshotCommand(ctx context.Context, rep repo.RepositoryWriter) error {
 	return errors.Errorf("encountered %v errors:\n%v", len(finalErrors), strings.Join(finalErrors, "\n"))
 }
 
+func getTags(tagStrings []string) (map[string]string, error) {
+	numberOfPartsInTagString := 2
+	// tagKeyPrefix is the prefix for user defined tag keys.
+	tagKeyPrefix := "tag:"
+
+	tags := map[string]string{}
+
+	for _, tagkv := range tagStrings {
+		parts := strings.SplitN(tagkv, ":", numberOfPartsInTagString)
+		if len(parts) != numberOfPartsInTagString {
+			return nil, errors.New("Invalid tag format. Requires <key>:<value>")
+		}
+
+		key := tagKeyPrefix + parts[0]
+		if _, ok := tags[key]; ok {
+			return nil, errors.Errorf("Duplicate tag <key> found. (%s)", parts[0])
+		}
+
+		tags[key] = parts[1]
+	}
+
+	return tags, nil
+}
+
 func validateStartEndTime(st, et string) error {
 	startTime, err := parseTimestamp(st)
 	if err != nil {
@@ -164,7 +194,7 @@ func startTimeAfterEndTime(startTime, endTime time.Time) bool {
 		startTime.After(endTime)
 }
 
-func snapshotSingleSource(ctx context.Context, rep repo.RepositoryWriter, u *snapshotfs.Uploader, sourceInfo snapshot.SourceInfo) error {
+func snapshotSingleSource(ctx context.Context, rep repo.RepositoryWriter, u *snapshotfs.Uploader, sourceInfo snapshot.SourceInfo, tags map[string]string) error {
 	log(ctx).Infof("Snapshotting %v ...", sourceInfo)
 
 	var (
@@ -207,6 +237,7 @@ func snapshotSingleSource(ctx context.Context, rep repo.RepositoryWriter, u *sna
 	}
 
 	manifest.Description = *snapshotCreateDescription
+	manifest.Tags = tags
 	startTimeOverride, _ := parseTimestamp(*snapshotCreateStartTime)
 	endTimeOverride, _ := parseTimestamp(*snapshotCreateEndTime)
 
diff --git a/cli/command_snapshot_list.go b/cli/command_snapshot_list.go
index d71a16166..6daaf9b2d 100644
--- a/cli/command_snapshot_list.go
+++ b/cli/command_snapshot_list.go
@@ -33,11 +33,12 @@ var (
 	snapshotListShowIdentical        = snapshotListCommand.Flag("show-identical", "Show identical snapshots").Short('l').Bool()
 	snapshotListShowAll              = snapshotListCommand.Flag("all", "Show all shapshots (not just current username/host)").Short('a').Bool()
 	maxResultsPerPath                = snapshotListCommand.Flag("max-results", "Maximum number of entries per source.").Short('n').Int()
+	snapshotListTags                 = snapshotListCommand.Flag("tags", "Tag filters to apply on the list items. Must be provided in the <key>:<value> format.").Strings()
 )
 
-func findSnapshotsForSource(ctx context.Context, rep repo.Repository, sourceInfo snapshot.SourceInfo) (manifestIDs []manifest.ID, relPath string, err error) {
+func findSnapshotsForSource(ctx context.Context, rep repo.Repository, sourceInfo snapshot.SourceInfo, tags map[string]string) (manifestIDs []manifest.ID, relPath string, err error) {
 	for len(sourceInfo.Path) > 0 {
-		list, err := snapshot.ListSnapshotManifests(ctx, rep, &sourceInfo)
+		list, err := snapshot.ListSnapshotManifests(ctx, rep, &sourceInfo, tags)
 		if err != nil {
 			return nil, "", errors.Wrapf(err, "error listing manifests for %v", sourceInfo)
 		}
@@ -65,9 +66,9 @@ func findSnapshotsForSource(ctx context.Context, rep repo.Repository, sourceInfo
 	return nil, "", nil
 }
 
-func findManifestIDs(ctx context.Context, rep repo.Repository, source string) ([]manifest.ID, string, error) {
+func findManifestIDs(ctx context.Context, rep repo.Repository, source string, tags map[string]string) ([]manifest.ID, string, error) {
 	if source == "" {
-		man, err := snapshot.ListSnapshotManifests(ctx, rep, nil)
+		man, err := snapshot.ListSnapshotManifests(ctx, rep, nil, tags)
 		return man, "", errors.Wrap(err, "error listing all snapshot manifests")
 	}
 
@@ -76,7 +77,7 @@ func findManifestIDs(ctx context.Context, rep repo.Repository, source string) ([
 		return nil, "", errors.Errorf("invalid directory: '%s': %s", source, err)
 	}
 
-	manifestIDs, relPath, err := findSnapshotsForSource(ctx, rep, si)
+	manifestIDs, relPath, err := findSnapshotsForSource(ctx, rep, si, tags)
 	if relPath != "" {
 		relPath = "/" + relPath
 	}
@@ -90,7 +91,12 @@ func runSnapshotsCommand(ctx context.Context, rep repo.Repository) error {
 	jl.begin()
 	defer jl.end()
 
-	manifestIDs, relPath, err := findManifestIDs(ctx, rep, *snapshotListPath)
+	tags, err := getTags(*snapshotListTags)
+	if err != nil {
+		return err
+	}
+
+	manifestIDs, relPath, err := findManifestIDs(ctx, rep, *snapshotListPath, tags)
 	if err != nil {
 		return err
 	}
diff --git a/cli/command_snapshot_migrate.go b/cli/command_snapshot_migrate.go
index 7fa69b88d..972ae1f4a 100644
--- a/cli/command_snapshot_migrate.go
+++ b/cli/command_snapshot_migrate.go
@@ -197,7 +197,7 @@ func findPreviousSnapshotManifestWithStartTime(ctx context.Context, rep repo.Rep
 }
 
 func migrateSingleSource(ctx context.Context, uploader *snapshotfs.Uploader, sourceRepo repo.Repository, destRepo repo.RepositoryWriter, s snapshot.SourceInfo) error {
-	manifests, err := snapshot.ListSnapshotManifests(ctx, sourceRepo, &s)
+	manifests, err := snapshot.ListSnapshotManifests(ctx, sourceRepo, &s, nil)
 	if err != nil {
 		return errors.Wrapf(err, "error listing snapshot manifests for %v", s)
 	}
diff --git a/cli/command_snapshot_verify.go b/cli/command_snapshot_verify.go
index 606bb97af..1f8f7e3da 100644
--- a/cli/command_snapshot_verify.go
+++ b/cli/command_snapshot_verify.go
@@ -267,7 +267,7 @@ func loadSourceManifests(ctx context.Context, rep repo.Repository, sources []str
 	var manifestIDs []manifest.ID
 
 	if len(sources)+len(*verifyCommandDirObjectIDs)+len(*verifyCommandFileObjectIDs) == 0 {
-		man, err := snapshot.ListSnapshotManifests(ctx, rep, nil)
+		man, err := snapshot.ListSnapshotManifests(ctx, rep, nil, nil)
 		if err != nil {
 			return nil, errors.Wrap(err, "unable to list snapshot manifests")
 		}
@@ -279,7 +279,7 @@ func loadSourceManifests(ctx context.Context, rep repo.Repository, sources []str
 			if err != nil {
 				return nil, errors.Wrapf(err, "error parsing %q", srcStr)
 			}
-			man, err := snapshot.ListSnapshotManifests(ctx, rep, &src)
+			man, err := snapshot.ListSnapshotManifests(ctx, rep, &src, nil)
 			if err != nil {
 				return nil, errors.Wrapf(err, "unable to list snapshot manifests for %v", src)
 			}
diff --git a/internal/server/api_snapshots.go b/internal/server/api_snapshots.go
index cf2d4b994..c60a9c955 100644
--- a/internal/server/api_snapshots.go
+++ b/internal/server/api_snapshots.go
@@ -11,7 +11,7 @@ import (
 )
 
 func (s *Server) handleSnapshotList(ctx context.Context, r *http.Request, body []byte) (interface{}, *apiError) {
-	manifestIDs, err := snapshot.ListSnapshotManifests(ctx, s.rep, nil)
+	manifestIDs, err := snapshot.ListSnapshotManifests(ctx, s.rep, nil, nil)
 	if err != nil {
 		return nil, internalServerError(err)
 	}
diff --git a/snapshot/manager.go b/snapshot/manager.go
index 645fe93bd..84d839d4b 100644
--- a/snapshot/manager.go
+++ b/snapshot/manager.go
@@ -127,7 +127,17 @@ func SaveSnapshot(ctx context.Context, rep repo.RepositoryWriter, man *Manifest)
 	// to write previous ID in JSON.
 	man.ID = ""
 
-	id, err := rep.PutManifest(ctx, sourceInfoToLabels(man.Source), man)
+	labels := sourceInfoToLabels(man.Source)
+
+	for key, value := range man.Tags {
+		if _, ok := labels[key]; ok {
+			return "", errors.Errorf("Invalid or duplicate tag <key> found in snapshot. (%s)", key)
+		}
+
+		labels[key] = value
+	}
+
+	id, err := rep.PutManifest(ctx, labels, man)
 	if err != nil {
 		return "", errors.Wrap(err, "error putting manifest")
 	}
@@ -175,7 +185,7 @@ func LoadSnapshots(ctx context.Context, rep repo.Repository, manifestIDs []manif
 }
 
 // ListSnapshotManifests returns the list of snapshot manifests for a given source or all sources if nil.
-func ListSnapshotManifests(ctx context.Context, rep repo.Repository, src *SourceInfo) ([]manifest.ID, error) {
+func ListSnapshotManifests(ctx context.Context, rep repo.Repository, src *SourceInfo, tags map[string]string) ([]manifest.ID, error) {
 	labels := map[string]string{
 		typeKey: ManifestType,
 	}
@@ -184,6 +194,10 @@ func ListSnapshotManifests(ctx context.Context, rep repo.Repository, src *Source
 		labels = sourceInfoToLabels(*src)
 	}
 
+	for key, value := range tags {
+		labels[key] = value
+	}
+
 	entries, err := rep.FindManifests(ctx, labels)
 	if err != nil {
 		return nil, errors.Wrap(err, "unable to find manifest entries")
@@ -194,7 +208,7 @@ func ListSnapshotManifests(ctx context.Context, rep repo.Repository, src *Source
 
 // FindSnapshotsByRootObjectID returns the list of matching snapshots for a given rootID.
 func FindSnapshotsByRootObjectID(ctx context.Context, rep repo.Repository, rootID object.ID) ([]*Manifest, error) {
-	ids, err := ListSnapshotManifests(ctx, rep, nil)
+	ids, err := ListSnapshotManifests(ctx, rep, nil, nil)
 	if err != nil {
 		return nil, errors.Wrap(err, "error listing snapshot manifests")
 	}
diff --git a/snapshot/manifest.go b/snapshot/manifest.go
index 8ceacc214..34d3dc4b9 100644
--- a/snapshot/manifest.go
+++ b/snapshot/manifest.go
@@ -28,6 +28,8 @@ type Manifest struct {
 	RootEntry *DirEntry `json:"rootEntry"`
 
 	RetentionReasons []string `json:"-"`
+
+	Tags map[string]string `json:"tags,omitempty"`
 }
 
 // EntryType is a type of a filesystem entry.
diff --git a/snapshot/snapshot_test.go b/snapshot/snapshot_test.go
index edfc5c8d4..ab8e6b1f6 100644
--- a/snapshot/snapshot_test.go
+++ b/snapshot/snapshot_test.go
@@ -81,7 +81,7 @@ func TestSnapshotsAPI(t *testing.T) {
 func verifySnapshotManifestIDs(t *testing.T, rep repo.Repository, src *snapshot.SourceInfo, expected []manifest.ID) {
 	t.Helper()
 
-	res, err := snapshot.ListSnapshotManifests(testlogging.Context(t), rep, src)
+	res, err := snapshot.ListSnapshotManifests(testlogging.Context(t), rep, src, nil)
 	if err != nil {
 		t.Errorf("error listing snapshot manifests: %v", err)
 	}
diff --git a/snapshot/snapshotgc/gc.go b/snapshot/snapshotgc/gc.go
index 778e20a45..a05cb5d56 100644
--- a/snapshot/snapshotgc/gc.go
+++ b/snapshot/snapshotgc/gc.go
@@ -27,7 +27,7 @@ func oidOf(entry fs.Entry) object.ID {
 }
 
 func findInUseContentIDs(ctx context.Context, rep repo.Repository, used *sync.Map) error {
-	ids, err := snapshot.ListSnapshotManifests(ctx, rep, nil)
+	ids, err := snapshot.ListSnapshotManifests(ctx, rep, nil, nil)
 	if err != nil {
 		return errors.Wrap(err, "unable to list snapshot manifest IDs")
 	}
diff --git a/tests/end_to_end_test/snapshot_create_test.go b/tests/end_to_end_test/snapshot_create_test.go
index 7106a951a..3db1fb241 100644
--- a/tests/end_to_end_test/snapshot_create_test.go
+++ b/tests/end_to_end_test/snapshot_create_test.go
@@ -81,6 +81,51 @@ func TestSnapshotCreate(t *testing.T) {
 	}
 }
 
+func TestTagging(t *testing.T) {
+	t.Parallel()
+
+	e := testenv.NewCLITest(t)
+
+	defer e.RunAndExpectSuccess(t, "repo", "disconnect")
+
+	e.RunAndExpectSuccess(t, "repo", "create", "filesystem", "--path", e.RepoDir)
+	e.RunAndExpectSuccess(t, "snapshot", "create", sharedTestDataDir1, "--tags", "testkey1:testkey2")
+	e.RunAndExpectSuccess(t, "snapshot", "create", sharedTestDataDir1)
+
+	var manifests []snapshot.Manifest
+
+	mustParseJSONLines(t, e.RunAndExpectSuccess(t, "snapshot", "list", "-a", "--json"), &manifests)
+
+	if got, want := len(manifests), 2; got != want {
+		t.Fatalf("unexpected number of snapshots %v want %v", got, want)
+	}
+
+	mustParseJSONLines(t, e.RunAndExpectSuccess(t, "snapshot", "list", "-a", "--tags", "testkey1:testkey2", "--json"), &manifests)
+
+	if got, want := len(manifests), 1; got != want {
+		t.Fatalf("unexpected number of snapshots %v want %v", got, want)
+	}
+}
+
+func TestTaggingBadTags(t *testing.T) {
+	t.Parallel()
+
+	e := testenv.NewCLITest(t)
+
+	defer e.RunAndExpectSuccess(t, "repo", "disconnect")
+
+	e.RunAndExpectSuccess(t, "repo", "create", "filesystem", "--path", e.RepoDir)
+
+	for _, tc := range [][]string{
+		{"--tags", "testkey1:testkey2", "--tags", "testkey1:testkey2"},
+		{"--tags", "badtag"},
+	} {
+		args := []string{"snapshot", "create", sharedTestDataDir1}
+		args = append(args, tc...)
+		e.RunAndExpectFailure(t, args...)
+	}
+}
+
 func TestStartTimeOverride(t *testing.T) {
 	t.Parallel()