Events in separate etcd

cmd/kube-apiserver/app/server.go

@@ -95,6 +95,7 @@ type APIServer struct {
 	AdmissionControlConfigFile string
 	EtcdServerList             []string
 	EtcdConfigFile             string
+	EtcdServersOverrides       []string
 	EtcdPathPrefix             string
 	CorsAllowedOriginList      []string
 	AllowPrivileged            bool
@@ -211,6 +212,7 @@ func (s *APIServer) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&s.AdmissionControlConfigFile, "admission-control-config-file", s.AdmissionControlConfigFile, "File with admission control configuration.")
 	fs.StringSliceVar(&s.EtcdServerList, "etcd-servers", s.EtcdServerList, "List of etcd servers to watch (http://ip:port), comma separated. Mutually exclusive with -etcd-config")
 	fs.StringVar(&s.EtcdConfigFile, "etcd-config", s.EtcdConfigFile, "The config file for the etcd client. Mutually exclusive with -etcd-servers.")
+	fs.StringSliceVar(&s.EtcdServersOverrides, "etcd-servers-overrides", s.EtcdServersOverrides, "Per-resource etcd servers overrides, comma separated. The individual override format: group/resource#servers, where servers are http://ip:port, semicolon separated.")
 	fs.StringVar(&s.EtcdPathPrefix, "etcd-prefix", s.EtcdPathPrefix, "The prefix for all resource paths in etcd.")
 	fs.StringSliceVar(&s.CorsAllowedOriginList, "cors-allowed-origins", s.CorsAllowedOriginList, "List of allowed origins for CORS, comma separated.  An allowed origin can be a regular expression to support subdomain matching.  If this list is empty CORS will not be enabled.")
 	fs.BoolVar(&s.AllowPrivileged, "allow-privileged", s.AllowPrivileged, "If true, allow privileged containers.")
@@ -253,6 +255,8 @@ func (s *APIServer) verifyClusterIPFlags() {
 	}
 }
 
+type newEtcdFunc func(string, []string, meta.VersionInterfacesFunc, string, string) (storage.Interface, error)
+
 func newEtcd(etcdConfigFile string, etcdServerList []string, interfacesFunc meta.VersionInterfacesFunc, storageVersion, pathPrefix string) (etcdStorage storage.Interface, err error) {
 	if storageVersion == "" {
 		return etcdStorage, fmt.Errorf("storageVersion is required to create a etcd storage")
@@ -294,6 +298,45 @@ func generateStorageVersionMap(legacyVersion string, storageVersions string) map
 	return storageVersionMap
 }
 
+// parse the value of --etcd-servers-overrides and update given storageDestinations.
+func updateEtcdOverrides(overrides []string, storageVersions map[string]string, prefix string, storageDestinations *master.StorageDestinations, newEtcdFn newEtcdFunc) {
+	if len(overrides) == 0 {
+		return
+	}
+	for _, override := range overrides {
+		tokens := strings.Split(override, "#")
+		if len(tokens) != 2 {
+			glog.Errorf("invalid value of etcd server overrides: %s", override)
+			continue
+		}
+
+		apiresource := strings.Split(tokens[0], "/")
+		if len(apiresource) != 2 {
+			glog.Errorf("invalid resource definition: %s", tokens[0])
+		}
+		group := apiresource[0]
+		resource := apiresource[1]
+
+		apigroup, err := latest.Group(group)
+		if err != nil {
+			glog.Errorf("invalid api group %s: %v", group, err)
+			continue
+		}
+		if _, found := storageVersions[apigroup.Group]; !found {
+			glog.Errorf("Couldn't find the storage version for group %s", apigroup.Group)
+			continue
+		}
+
+		servers := strings.Split(tokens[1], ";")
+		etcdOverrideStorage, err := newEtcdFn("", servers, apigroup.InterfacesFor, storageVersions[apigroup.Group], prefix)
+		if err != nil {
+			glog.Fatalf("Invalid storage version or misconfigured etcd for %s: %v", tokens[0], err)
+		}
+
+		storageDestinations.AddStorageOverride(group, resource, etcdOverrideStorage)
+	}
+}
+
 // Run runs the specified APIServer.  This should never exit.
 func (s *APIServer) Run(_ []string) error {
 	s.verifyClusterIPFlags()
@@ -369,6 +412,8 @@ func (s *APIServer) Run(_ []string) error {
 		return err
 	}
 
+	storageDestinations := master.NewStorageDestinations()
+
 	storageVersions := generateStorageVersionMap(s.DeprecatedStorageVersion, s.StorageVersions)
 	if _, found := storageVersions[legacyV1Group.Group]; !found {
 		glog.Fatalf("Couldn't find the storage version for group: %q in storageVersions: %v", legacyV1Group.Group, storageVersions)
@@ -377,8 +422,8 @@ func (s *APIServer) Run(_ []string) error {
 	if err != nil {
 		glog.Fatalf("Invalid storage version or misconfigured etcd: %v", err)
 	}
+	storageDestinations.AddAPIGroup("", etcdStorage)
 
-	var expEtcdStorage storage.Interface
 	if enableExp {
 		expGroup, err := latest.Group("experimental")
 		if err != nil {
@@ -387,12 +432,15 @@ func (s *APIServer) Run(_ []string) error {
 		if _, found := storageVersions[expGroup.Group]; !found {
 			glog.Fatalf("Couldn't find the storage version for group: %q in storageVersions: %v", expGroup.Group, storageVersions)
 		}
-		expEtcdStorage, err = newEtcd(s.EtcdConfigFile, s.EtcdServerList, expGroup.InterfacesFor, storageVersions[expGroup.Group], s.EtcdPathPrefix)
+		expEtcdStorage, err := newEtcd(s.EtcdConfigFile, s.EtcdServerList, expGroup.InterfacesFor, storageVersions[expGroup.Group], s.EtcdPathPrefix)
 		if err != nil {
 			glog.Fatalf("Invalid experimental storage version or misconfigured etcd: %v", err)
 		}
+		storageDestinations.AddAPIGroup("experimental", expEtcdStorage)
 	}
 
+	updateEtcdOverrides(s.EtcdServersOverrides, storageVersions, s.EtcdPathPrefix, &storageDestinations, newEtcd)
+
 	n := s.ServiceClusterIPRange
 
 	// Default to the private server key for service account token signing
@@ -460,10 +508,8 @@ func (s *APIServer) Run(_ []string) error {
 		}
 	}
 	config := &master.Config{
-		DatabaseStorage:    etcdStorage,
-		ExpDatabaseStorage: expEtcdStorage,
-		StorageVersions:    storageVersions,
-
+		StorageDestinations:    storageDestinations,
+		StorageVersions:        storageVersions,
 		EventTTL:               s.EventTTL,
 		KubeletClient:          kubeletClient,
 		ServiceClusterIPRange:  &n,