Merge pull request #20358 from nikhiljindal/federatedProxy

Automatic merge from submit-queue

federated api servers: Adding a discovery summarizer server

Ref https://github.com/kubernetes/kubernetes/issues/20229

The server reads a config file to get the list of federated servers and then surfaces the combined discovery information at `/api` and `/apis`.


cc @kubernetes/sig-api-machinery @lavalamp

<!-- Reviewable:start -->
---
This change is [<img src="http://reviewable.k8s.io/review_button.svg" height="35" align="absmiddle" alt="Reviewable"/>](http://reviewable.k8s.io/reviews/kubernetes/kubernetes/20358)
<!-- Reviewable:end -->

cmd/kubernetes-discovery/config.json

@@ -0,0 +1,16 @@
+{
+        "servers": [
+                {
+                        "serverAddress": "http://127.0.0.1:8083",
+                        "groupVersionDiscoveryPaths": [
+                                {
+                                        "path": "/apis"
+                                },
+                                {
+                                        "path": "/api",
+                                        "isLegacy": true
+                                }
+                        ]
+                }
+        ]
+}

cmd/kubernetes-discovery/discoverysummarizer/apis/config/v1alpha1/types.go

@@ -0,0 +1,43 @@
+/*
+Copyright 2016 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1alpha1
+
+// List of servers from which group versions should be summarized.
+// This is used to represent the structure of the config file passed to discovery summarizer server.
+type FederatedServerList struct {
+	Servers []FederatedServer `json:"servers"`
+}
+
+// Information about each individual server, whose group versions needs to be summarized.
+type FederatedServer struct {
+	// The address that summarizer can reach to get discovery information from the server.
+	// This can be hostname, hostname:port, IP or IP:port.
+	ServerAddress string `json:"serverAddress"`
+	// The list of paths where server exposes group version discovery information.
+	// Summarizer will use these paths to figure out group versions supported by this server.
+	GroupVersionDiscoveryPaths []GroupVersionDiscoveryPath `json:"groupVersionDiscoveryPaths"`
+}
+
+// Information about each group version discovery path that needs to be summarized.
+type GroupVersionDiscoveryPath struct {
+	// Path where the server exposes the discovery API to surface the group versions that it supports.
+	Path string `json:"path"`
+
+	// True if the path is for legacy group version.
+	// (i.e the path returns unversioned.APIVersions instead of unversioned.APIGroupList)
+	IsLegacy bool `json:"isLegacy"`
+}

cmd/kubernetes-discovery/discoverysummarizer/discoverysummarizer.go

@@ -0,0 +1,192 @@
+/*
+Copyright 2016 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package discoverysummarizer
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+
+	config "k8s.io/kubernetes/cmd/kubernetes-discovery/discoverysummarizer/apis/config/v1alpha1"
+	"k8s.io/kubernetes/pkg/api/unversioned"
+)
+
+type DiscoverySummarizer interface {
+	Run(port string) error
+}
+
+type discoverySummarizerServer struct {
+	// The list of servers as read from the config file.
+	serverList         config.FederatedServerList
+	groupVersionPaths  map[string][]string
+	legacyVersionPaths map[string][]string
+}
+
+// Ensure that discoverySummarizerServer implements DiscoverySummarizer interface.
+var _ DiscoverySummarizer = &discoverySummarizerServer{}
+
+// Creates a server to summarize all group versions
+// supported by the servers mentioned in the given config file.
+// Call Run() to bring up the server.
+func NewDiscoverySummarizer(configFilePath string) (DiscoverySummarizer, error) {
+	file, err := ioutil.ReadFile(configFilePath)
+	if err != nil {
+		return nil, fmt.Errorf("Error in reading config file: %v\n", err)
+	}
+	ds := discoverySummarizerServer{
+		groupVersionPaths:  map[string][]string{},
+		legacyVersionPaths: map[string][]string{},
+	}
+	err = json.Unmarshal(file, &ds.serverList)
+	if err != nil {
+		return nil, fmt.Errorf("Error in marshalling config file to json: %v\n", err)
+	}
+
+	for _, server := range ds.serverList.Servers {
+		for _, groupVersionPath := range server.GroupVersionDiscoveryPaths {
+			if groupVersionPath.IsLegacy {
+				ds.legacyVersionPaths[groupVersionPath.Path] = append(ds.legacyVersionPaths[groupVersionPath.Path], server.ServerAddress)
+			} else {
+				ds.groupVersionPaths[groupVersionPath.Path] = append(ds.groupVersionPaths[groupVersionPath.Path], server.ServerAddress)
+			}
+		}
+	}
+	return &ds, nil
+}
+
+// Brings up the server at the given port.
+// TODO: Add HTTPS support.
+func (ds *discoverySummarizerServer) Run(port string) error {
+	http.HandleFunc("/", ds.indexHandler)
+	// Register a handler for all paths.
+	for path := range ds.groupVersionPaths {
+		p := path
+		fmt.Printf("setting up a handler for %s\n", p)
+		http.HandleFunc(p, ds.summarizeGroupVersionsHandler(p))
+	}
+	for path := range ds.legacyVersionPaths {
+		p := path
+		fmt.Printf("setting up a handler for %s\n", p)
+		http.HandleFunc(p, ds.summarizeLegacyVersionsHandler(p))
+	}
+	fmt.Printf("Server running on port %s\n", port)
+	return http.ListenAndServe(":"+port, nil)
+}
+
+// Handler for "/"
+func (ds *discoverySummarizerServer) indexHandler(w http.ResponseWriter, r *http.Request) {
+	if r.URL.Path != "/" {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	w.Write([]byte("Success"))
+}
+
+// Handler for group versions summarizer.
+func (ds *discoverySummarizerServer) summarizeGroupVersionsHandler(path string) func(http.ResponseWriter, *http.Request) {
+	return func(w http.ResponseWriter, r *http.Request) {
+		var apiGroupList *unversioned.APIGroupList
+		// TODO: We can cache and parallelize the calls to all servers.
+		for _, serverAddress := range ds.groupVersionPaths[path] {
+			groupList, err := ds.getAPIGroupList(serverAddress + path)
+			if err != nil {
+				ds.writeErr(http.StatusBadGateway, err, w)
+				return
+			}
+			if apiGroupList == nil {
+				apiGroupList = &unversioned.APIGroupList{}
+				*apiGroupList = *groupList
+			} else {
+				apiGroupList.Groups = append(apiGroupList.Groups, groupList.Groups...)
+			}
+		}
+		ds.writeRawJSON(http.StatusOK, *apiGroupList, w)
+		return
+	}
+}
+
+// Handler for legacy versions summarizer.
+func (ds *discoverySummarizerServer) summarizeLegacyVersionsHandler(path string) func(http.ResponseWriter, *http.Request) {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if len(ds.legacyVersionPaths[path]) > 1 {
+			err := fmt.Errorf("invalid multiple servers serving legacy group %v", ds.legacyVersionPaths[path])
+			ds.writeErr(http.StatusInternalServerError, err, w)
+		}
+		serverAddress := ds.legacyVersionPaths[path][0]
+		apiVersions, err := ds.getAPIVersions(serverAddress + path)
+		if err != nil {
+			ds.writeErr(http.StatusBadGateway, err, w)
+			return
+		}
+		ds.writeRawJSON(http.StatusOK, apiVersions, w)
+		return
+	}
+}
+
+func (ds *discoverySummarizerServer) getAPIGroupList(serverAddress string) (*unversioned.APIGroupList, error) {
+	response, err := http.Get(serverAddress)
+	if err != nil {
+		return nil, fmt.Errorf("Error in fetching %s: %v", serverAddress, err)
+	}
+	defer response.Body.Close()
+	contents, err := ioutil.ReadAll(response.Body)
+	if err != nil {
+		return nil, fmt.Errorf("Error reading response from %s: %v", serverAddress, err)
+	}
+	var apiGroupList unversioned.APIGroupList
+	err = json.Unmarshal(contents, &apiGroupList)
+	if err != nil {
+		return nil, fmt.Errorf("Error in unmarshalling response from server %s: %v", serverAddress, err)
+	}
+	return &apiGroupList, nil
+}
+
+func (ds *discoverySummarizerServer) getAPIVersions(serverAddress string) (*unversioned.APIVersions, error) {
+	response, err := http.Get(serverAddress)
+	if err != nil {
+		return nil, fmt.Errorf("Error in fetching %s: %v", serverAddress, err)
+	}
+	defer response.Body.Close()
+	contents, err := ioutil.ReadAll(response.Body)
+	if err != nil {
+		return nil, fmt.Errorf("Error reading response from %s: %v", serverAddress, err)
+	}
+	var apiVersions unversioned.APIVersions
+	err = json.Unmarshal(contents, &apiVersions)
+	if err != nil {
+		return nil, fmt.Errorf("Error in unmarshalling response from server %s: %v", serverAddress, err)
+	}
+	return &apiVersions, nil
+}
+
+// TODO: Pass a runtime.Object here instead of interface{} and use the encoding/decoding stack from kubernetes apiserver.
+func (ds *discoverySummarizerServer) writeRawJSON(statusCode int, object interface{}, w http.ResponseWriter) {
+	output, err := json.MarshalIndent(object, "", "  ")
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(statusCode)
+	w.Write(output)
+}
+
+func (ds *discoverySummarizerServer) writeErr(statusCode int, err error, w http.ResponseWriter) {
+	http.Error(w, err.Error(), statusCode)
+}

cmd/kubernetes-discovery/discoverysummarizer/discoverysummarizer_test.go

@@ -0,0 +1,112 @@
+/*
+Copyright 2016 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package discoverysummarizer
+
+import (
+	"fmt"
+	"net/http"
+	"testing"
+	"time"
+
+	"k8s.io/kubernetes/examples/apiserver"
+)
+
+func waitForServerUp(serverURL string) error {
+	for start := time.Now(); time.Since(start) < time.Minute; time.Sleep(5 * time.Second) {
+		_, err := http.Get(serverURL)
+		if err == nil {
+			return nil
+		}
+	}
+	return fmt.Errorf("waiting for server timed out")
+}
+
+func testResponse(t *testing.T, serverURL, path string, expectedStatusCode int) {
+	response, err := http.Get(serverURL + path)
+	if err != nil {
+		t.Errorf("unexpected error in GET %s: %v", path, err)
+	}
+	if response.StatusCode != expectedStatusCode {
+		t.Errorf("unexpected status code: %v, expected: %v", response.StatusCode, expectedStatusCode)
+	}
+}
+
+func runDiscoverySummarizer(t *testing.T) string {
+	configFilePath := "../config.json"
+	port := "9090"
+	serverURL := "http://localhost:" + port
+	s, err := NewDiscoverySummarizer(configFilePath)
+	if err != nil {
+		t.Errorf("unexpected error: %v\n", err)
+	}
+	go func() {
+		if err := s.Run(port); err != nil {
+			t.Fatalf("error in bringing up the server: %v", err)
+		}
+	}()
+	if err := waitForServerUp(serverURL); err != nil {
+		t.Fatalf("%v", err)
+	}
+	return serverURL
+}
+
+func runAPIServer(t *testing.T) string {
+	serverRunOptions := apiserver.NewServerRunOptions()
+	// Change the port, because otherwise it will fail if examples/apiserver/apiserver_test and this are run in parallel.
+	serverRunOptions.InsecurePort = 8083
+	go func() {
+		if err := apiserver.Run(serverRunOptions); err != nil {
+			t.Fatalf("Error in bringing up the example apiserver: %v", err)
+		}
+	}()
+
+	serverURL := fmt.Sprintf("http://localhost:%d", serverRunOptions.InsecurePort)
+	if err := waitForServerUp(serverURL); err != nil {
+		t.Fatalf("%v", err)
+	}
+	return serverURL
+}
+
+// Runs a discovery summarizer server and tests that all endpoints work as expected.
+func TestRunDiscoverySummarizer(t *testing.T) {
+	discoveryURL := runDiscoverySummarizer(t)
+
+	// Test /api path.
+	// There is no server running at that URL, so we will get a 500.
+	testResponse(t, discoveryURL, "/api", http.StatusBadGateway)
+
+	// Test /apis path.
+	// There is no server running at that URL, so we will get a 500.
+	testResponse(t, discoveryURL, "/apis", http.StatusBadGateway)
+
+	// Test a random path, which should give a 404.
+	testResponse(t, discoveryURL, "/randomPath", http.StatusNotFound)
+
+	// Run the APIServer now to test the good case.
+	runAPIServer(t)
+
+	// Test /api path.
+	// There is no server running at that URL, so we will get a 500.
+	testResponse(t, discoveryURL, "/api", http.StatusOK)
+
+	// Test /apis path.
+	// There is no server running at that URL, so we will get a 500.
+	testResponse(t, discoveryURL, "/apis", http.StatusOK)
+
+	// Test a random path, which should give a 404.
+	testResponse(t, discoveryURL, "/randomPath", http.StatusNotFound)
+}

cmd/kubernetes-discovery/discoverysummarizer/doc.go

@@ -0,0 +1,20 @@
+/*
+Copyright 2016 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package discoverysummarizer contains code for the discovery summarizer
+// (program to summarize discovery information from all federated api servers)
+// as per https://github.com/kubernetes/kubernetes/blob/master/docs/proposals/federated-api-servers.md
+package discoverysummarizer

cmd/kubernetes-discovery/main.go

@@ -0,0 +1,37 @@
+/*
+Copyright 2016 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"k8s.io/kubernetes/cmd/kubernetes-discovery/discoverysummarizer"
+
+	"github.com/golang/glog"
+)
+
+func main() {
+	// TODO: move them to flags.
+	configFilePath := "config.json"
+	port := "9090"
+	s, err := discoverysummarizer.NewDiscoverySummarizer(configFilePath)
+	if err != nil {
+		glog.Fatalf("%v\n", err)
+	}
+	err = s.Run(port)
+	if err != nil {
+		glog.Fatalf("%v\n", err)
+	}
+}

examples/apiserver/apiserver.go

@@ -37,7 +37,6 @@ const (
 	// Ports on which to run the server.
 	// Explicitly setting these to a different value than the default values, to prevent this from clashing with a local cluster.
 	InsecurePort = 8081
-	SecurePort   = 6444
 )
 
 func newStorageFactory() genericapiserver.StorageFactory {
@@ -53,7 +52,6 @@ func newStorageFactory() genericapiserver.StorageFactory {
 func NewServerRunOptions() *genericapiserver.ServerRunOptions {
 	serverOptions := genericapiserver.NewServerRunOptions()
 	serverOptions.InsecurePort = InsecurePort
-	serverOptions.SecurePort = SecurePort
 	return serverOptions
 }