diff --git a/cmd/containerd/command/config.go b/cmd/containerd/command/config.go
index 4b0f5bf7f..8bb917d35 100644
--- a/cmd/containerd/command/config.go
+++ b/cmd/containerd/command/config.go
@@ -24,6 +24,7 @@ import (
 	"github.com/containerd/containerd/defaults"
 	"github.com/containerd/containerd/images"
 	"github.com/containerd/containerd/pkg/timeout"
+	"github.com/containerd/containerd/plugin"
 	"github.com/containerd/containerd/services/server"
 	srvconfig "github.com/containerd/containerd/services/server/config"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
@@ -65,8 +66,8 @@ func outputConfig(ctx gocontext.Context, config *srvconfig.Config) error {
 	// for the time being, keep the defaultConfig's version set at 1 so that
 	// when a config without a version is loaded from disk and has no version
 	// set, we assume it's a v1 config.  But when generating new configs via
-	// this command, generate the v2 config
-	config.Version = 2
+	// this command, generate the max configuration version
+	config.Version = srvconfig.CurrentConfigVersion
 
 	return toml.NewEncoder(os.Stdout).Encode(config)
 }
@@ -99,16 +100,38 @@ var configCommand = cli.Command{
 				return outputConfig(ctx, config)
 			},
 		},
+		{
+			Name:  "migrate",
+			Usage: "Migrate the current configuration file to the latest version (does not migrate subconfig files)",
+			Action: func(context *cli.Context) error {
+				config := defaultConfig()
+				ctx := gocontext.Background()
+				if err := srvconfig.LoadConfig(ctx, context.GlobalString("config"), config); err != nil && !os.IsNotExist(err) {
+					return err
+				}
+
+				if config.Version < srvconfig.CurrentConfigVersion {
+					plugins := plugin.Graph(srvconfig.V2DisabledFilter(config.DisabledPlugins))
+					for _, p := range plugins {
+						if p.ConfigMigration != nil {
+							if err := p.ConfigMigration(ctx, config.Version, config.Plugins); err != nil {
+								return err
+							}
+						}
+					}
+				}
+
+				config.Version = srvconfig.CurrentConfigVersion
+
+				return toml.NewEncoder(os.Stdout).SetIndentTables(true).Encode(config)
+			},
+		},
 	},
 }
 
 func platformAgnosticDefaultConfig() *srvconfig.Config {
 	return &srvconfig.Config{
-		// see: https://github.com/containerd/containerd/blob/5c6ea7fdc1247939edaddb1eba62a94527418687/RELEASES.md#daemon-configuration
-		// this version MUST remain set to 1 until either there exists a means to
-		// override / configure the default at the containerd cli .. or when
-		// version 1 is no longer supported
-		Version: 1,
+		Version: srvconfig.CurrentConfigVersion,
 		Root:    defaults.DefaultRootDir,
 		State:   defaults.DefaultStateDir,
 		GRPC: srvconfig.GRPCConfig{
diff --git a/services/server/config/config.go b/services/server/config/config.go
index 59a44d777..0f395f040 100644
--- a/services/server/config/config.go
+++ b/services/server/config/config.go
@@ -14,6 +14,12 @@
    limitations under the License.
 */
 
+// config is the global configuration for containerd
+//
+// Version History
+// 1: Deprecated and removed in containerd 2.0
+// 2: Uses fully qualified plugin names
+// 3: Added support for migration and warning on unknown fields
 package config
 
 import (
@@ -34,6 +40,9 @@ import (
 	"github.com/containerd/log"
 )
 
+// CurrentConfigVersion is the max config version which is supported
+const CurrentConfigVersion = 3
+
 // NOTE: Any new map fields added also need to be handled in mergeConfig.
 
 // Config provides containerd configuration data for the server
@@ -100,17 +109,18 @@ func (c *Config) GetVersion() int {
 	return c.Version
 }
 
-// ValidateV2 validates the config for a v2 file
-func (c *Config) ValidateV2() error {
-	switch version := c.GetVersion(); version {
-	case 1:
+// ValidateVersion validates the config for a v2 file
+func (c *Config) ValidateVersion() error {
+	version := c.GetVersion()
+	if version == 1 {
 		return errors.New("containerd config version `1` is no longer supported since containerd v2.0, please switch to version `2`, " +
 			"see https://github.com/containerd/containerd/blob/main/docs/PLUGINS.md#version-header")
-	case 2:
-		// NOP
-	default:
-		return fmt.Errorf("expected containerd config version `2`, got `%d`", version)
 	}
+
+	if version > CurrentConfigVersion {
+		return fmt.Errorf("expected containerd config version equal to or less than `%d`, got `%d`", CurrentConfigVersion, version)
+	}
+
 	for _, p := range c.DisabledPlugins {
 		if !strings.HasPrefix(p, "io.containerd.") || len(strings.SplitN(p, ".", 4)) < 4 {
 			return fmt.Errorf("invalid disabled plugin URI %q expect io.containerd.x.vx", p)
@@ -253,7 +263,7 @@ func LoadConfig(ctx context.Context, path string, out *Config) error {
 		out.Imports = append(out.Imports, path)
 	}
 
-	err := out.ValidateV2()
+	err := out.ValidateVersion()
 	if err != nil {
 		return fmt.Errorf("failed to load TOML from %s: %w", path, err)
 	}
diff --git a/services/server/server.go b/services/server/server.go
index 46e56338d..298f7e4be 100644
--- a/services/server/server.go
+++ b/services/server/server.go
@@ -202,6 +202,16 @@ func New(ctx context.Context, config *srvconfig.Config) (*Server, error) {
 		required[r] = struct{}{}
 	}
 
+	if config.Version < srvconfig.CurrentConfigVersion {
+		for _, p := range plugins {
+			if p.ConfigMigration != nil {
+				if err := p.ConfigMigration(ctx, config.Version, config.Plugins); err != nil {
+					return nil, err
+				}
+			}
+		}
+	}
+
 	for _, p := range plugins {
 		id := p.URI()
 		log.G(ctx).WithField("type", p.Type).Infof("loading plugin %q...", id)
diff --git a/services/server/server_test.go b/services/server/server_test.go
index 82dd2dd6a..3c0faaf0f 100644
--- a/services/server/server_test.go
+++ b/services/server/server_test.go
@@ -17,8 +17,10 @@
 package server
 
 import (
+	"context"
 	"testing"
 
+	"github.com/containerd/containerd/plugin"
 	srvconfig "github.com/containerd/containerd/services/server/config"
 	"github.com/stretchr/testify/assert"
 )
@@ -53,3 +55,94 @@ func TestCreateTopLevelDirectoriesWithEmptyRootPath(t *testing.T) {
 	})
 	assert.EqualError(t, err, "root must be specified")
 }
+
+func TestMigration(t *testing.T) {
+	plugin.Reset()
+	defer plugin.Reset()
+
+	version := srvconfig.CurrentConfigVersion - 1
+
+	type testConfig struct {
+		Migrated    string `toml:"migrated"`
+		NotMigrated string `toml:"notmigrated"`
+	}
+
+	plugin.Register(&plugin.Registration{
+		Type:   "io.containerd.test",
+		ID:     "t1",
+		Config: &testConfig{},
+		InitFn: func(ic *plugin.InitContext) (interface{}, error) {
+			c, ok := ic.Config.(*testConfig)
+			if !ok {
+				t.Error("expected first plugin to have configuration")
+			} else {
+				if c.Migrated != "" {
+					t.Error("expected first plugin to have empty value for migrated config")
+				}
+				if c.NotMigrated != "don't migrate me" {
+					t.Errorf("expected first plugin does not have correct value for not migrated config: %q", c.NotMigrated)
+				}
+			}
+			return nil, nil
+		},
+	})
+	plugin.Register(&plugin.Registration{
+		Type: "io.containerd.new",
+		Requires: []plugin.Type{
+			"io.containerd.test", // Ensure this test runs second
+		},
+		ID:     "t2",
+		Config: &testConfig{},
+		InitFn: func(ic *plugin.InitContext) (interface{}, error) {
+			c, ok := ic.Config.(*testConfig)
+			if !ok {
+				t.Error("expected second plugin to have configuration")
+			} else {
+				if c.Migrated != "migrate me" {
+					t.Errorf("expected second plugin does not have correct value for migrated config: %q", c.Migrated)
+				}
+				if c.NotMigrated != "" {
+					t.Error("expected second plugin to have empty value for not migrated config")
+				}
+			}
+			return nil, nil
+		},
+		ConfigMigration: func(ctx context.Context, v int, plugins map[string]interface{}) error {
+			if v != version {
+				t.Errorf("unxpected version: %d", v)
+			}
+			t1, ok := plugins["io.containerd.test.t1"]
+			if !ok {
+				t.Error("plugin not set as expected")
+				return nil
+			}
+			conf, ok := t1.(map[string]interface{})
+			if !ok {
+				t.Errorf("unexpected config value: %v", t1)
+				return nil
+			}
+			newconf := map[string]interface{}{
+				"migrated": conf["migrated"],
+			}
+			delete(conf, "migrated")
+			plugins["io.containerd.new.t2"] = newconf
+
+			return nil
+		},
+	})
+
+	config := &srvconfig.Config{}
+	config.Version = version
+	config.Plugins = map[string]interface{}{
+		"io.containerd.test.t1": map[string]interface{}{
+			"migrated":    "migrate me",
+			"notmigrated": "don't migrate me",
+		},
+	}
+
+	ctx := context.Background()
+	_, err := New(ctx, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+}