Skip to content

Commit

Permalink
Add watch mechanism to apiserver
Browse files Browse the repository at this point in the history 
Implemented via HTTP and websocket. A test is present but this isn't
yet wired into anything.

Eventual purpose of this is to allow a scheduler to watch for new pods.
Or allow replication controller to watch for new items it controlls.
Generally, it'll be good to turn everything possible into a push instead
of a poll.
  • Loading branch information
@lavalamp
lavalamp committed Jul 18, 2014
1 parent af0ded7 commit eda30d4
Show file tree
Hide file tree
Showing 7 changed files with 535 additions and 29 deletions.
10 changes: 10 additions & 0 deletions pkg/api/types.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ package api

import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
"github.com/fsouza/go-dockerclient"
)

Expand Down Expand Up @@ -338,3 +339,12 @@ type ServerOpList struct {
JSONBase `yaml:",inline" json:",inline"`
Items []ServerOp `yaml:"items,omitempty" json:"items,omitempty"`
}

// WatchEvent objects are streamed from the api server in response to a watch request.
type WatchEvent struct {
// The type of the watch event; added, modified, or deleted.
Type watch.EventType

// An object which can be decoded via api.Decode
EmbeddedObject []byte
}
204 changes: 181 additions & 23 deletions pkg/apiserver/apiserver.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ limitations under the License.
package apiserver

import (
"encoding/json"
"fmt"
"io/ioutil"
"net/http"
Expand All @@ -25,12 +26,14 @@ import (
"strings"
"time"

"code.google.com/p/go.net/websocket"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/healthz"
"github.com/GoogleCloudPlatform/kubernetes/pkg/httplog"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
"github.com/golang/glog"
)

Expand Down Expand Up @@ -72,6 +75,13 @@ type RESTStorage interface {
Update(interface{}) (<-chan interface{}, error)
}

// ResourceWatcher should be implemented by all RESTStorage objects that
// want to offer the ability to watch for changes through the watch api.
type ResourceWatcher interface {
WatchAll() (watch.Interface, error)
WatchSingle(id string) (watch.Interface, error)
}

// WorkFunc is used to perform any time consuming work for an api call, after
// the input has been validated. Pass one of these to MakeAsync to create an
// appropriate return value for the Update, Delete, and Create methods.
Expand Down Expand Up @@ -136,13 +146,24 @@ func New(storage map[string]RESTStorage, prefix string) *APIServer {
s.mux.HandleFunc("/index.html", s.handleIndex)

// Handle both operations and operations/* with the same handler
opPrefix := path.Join(s.prefix, "operations")
s.mux.HandleFunc(opPrefix, s.handleOperationRequest)
s.mux.HandleFunc(opPrefix+"/", s.handleOperationRequest)
s.mux.HandleFunc(s.operationPrefix(), s.handleOperationRequest)
s.mux.HandleFunc(s.operationPrefix()+"/", s.handleOperationRequest)

s.mux.HandleFunc(s.watchPrefix()+"/", s.handleWatch)

s.mux.HandleFunc("/", s.notFound)

return s
}

func (s *APIServer) operationPrefix() string {
return path.Join(s.prefix, "operations")
}

func (s *APIServer) watchPrefix() string {
return path.Join(s.prefix, "watch")
}

func (server *APIServer) handleIndex(w http.ResponseWriter, req *http.Request) {
w.WriteHeader(http.StatusOK)
// TODO: serve this out of a file?
Expand Down Expand Up @@ -175,25 +196,25 @@ func (server *APIServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
// ServeREST handles requests to all our RESTStorage objects.
func (server *APIServer) ServeREST(w http.ResponseWriter, req *http.Request) {
if !strings.HasPrefix(req.URL.Path, server.prefix) {
server.notFound(req, w)
server.notFound(w, req)
return
}
requestParts := strings.Split(req.URL.Path[len(server.prefix):], "/")[1:]
if len(requestParts) < 1 {
server.notFound(req, w)
server.notFound(w, req)
return
}
storage := server.storage[requestParts[0]]
if storage == nil {
httplog.LogOf(w).Addf("'%v' has no storage object", requestParts[0])
server.notFound(req, w)
server.notFound(w, req)
return
}

server.handleREST(requestParts, req, w, storage)
}

func (server *APIServer) notFound(req *http.Request, w http.ResponseWriter) {
func (server *APIServer) notFound(w http.ResponseWriter, req *http.Request) {
w.WriteHeader(http.StatusNotFound)
fmt.Fprintf(w, "Not Found: %#v", req)
}
Expand Down Expand Up @@ -290,7 +311,7 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
case 2:
item, err := storage.Get(parts[1])
if IsNotFound(err) {
server.notFound(req, w)
server.notFound(w, req)
return
}
if err != nil {
Expand All @@ -299,11 +320,11 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
}
server.write(http.StatusOK, item, w)
default:
server.notFound(req, w)
server.notFound(w, req)
}
case "POST":
if len(parts) != 1 {
server.notFound(req, w)
server.notFound(w, req)
return
}
body, err := server.readBody(req)
Expand All @@ -313,7 +334,7 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
}
obj, err := storage.Extract(body)
if IsNotFound(err) {
server.notFound(req, w)
server.notFound(w, req)
return
}
if err != nil {
Expand All @@ -322,7 +343,7 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
}
out, err := storage.Create(obj)
if IsNotFound(err) {
server.notFound(req, w)
server.notFound(w, req)
return
}
if err != nil {
Expand All @@ -332,12 +353,12 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
server.finishReq(out, sync, timeout, w)
case "DELETE":
if len(parts) != 2 {
server.notFound(req, w)
server.notFound(w, req)
return
}
out, err := storage.Delete(parts[1])
if IsNotFound(err) {
server.notFound(req, w)
server.notFound(w, req)
return
}
if err != nil {
Expand All @@ -347,7 +368,7 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
server.finishReq(out, sync, timeout, w)
case "PUT":
if len(parts) != 2 {
server.notFound(req, w)
server.notFound(w, req)
return
}
body, err := server.readBody(req)
Expand All @@ -357,7 +378,7 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
}
obj, err := storage.Extract(body)
if IsNotFound(err) {
server.notFound(req, w)
server.notFound(w, req)
return
}
if err != nil {
Expand All @@ -366,7 +387,7 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
}
out, err := storage.Update(obj)
if IsNotFound(err) {
server.notFound(req, w)
server.notFound(w, req)
return
}
if err != nil {
Expand All @@ -375,24 +396,24 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
}
server.finishReq(out, sync, timeout, w)
default:
server.notFound(req, w)
server.notFound(w, req)
}
}

func (server *APIServer) handleOperationRequest(w http.ResponseWriter, req *http.Request) {
opPrefix := path.Join(server.prefix, "operations")
opPrefix := server.operationPrefix()
if !strings.HasPrefix(req.URL.Path, opPrefix) {
server.notFound(req, w)
server.notFound(w, req)
return
}
trimmed := strings.TrimLeft(req.URL.Path[len(opPrefix):], "/")
parts := strings.Split(trimmed, "/")
if len(parts) > 1 {
server.notFound(req, w)
server.notFound(w, req)
return
}
if req.Method != "GET" {
server.notFound(req, w)
server.notFound(w, req)
return
}
if len(parts) == 0 {
Expand All @@ -404,7 +425,7 @@ func (server *APIServer) handleOperationRequest(w http.ResponseWriter, req *http

op := server.ops.Get(parts[0])
if op == nil {
server.notFound(req, w)
server.notFound(w, req)
return
}

Expand All @@ -415,3 +436,140 @@ func (server *APIServer) handleOperationRequest(w http.ResponseWriter, req *http
server.write(http.StatusAccepted, obj, w)
}
}

func (server *APIServer) handleWatch(w http.ResponseWriter, req *http.Request) {
prefix := server.watchPrefix()
if !strings.HasPrefix(req.URL.Path, prefix) {
server.notFound(w, req)
return
}
parts := strings.Split(req.URL.Path[len(prefix):], "/")[1:]
if req.Method != "GET" || len(parts) < 1 {
server.notFound(w, req)
}
storage := server.storage[parts[0]]
if storage == nil {
server.notFound(w, req)
}
if watcher, ok := storage.(ResourceWatcher); ok {
var watching watch.Interface
var err error
if id := req.URL.Query().Get("id"); id != "" {
watching, err = watcher.WatchSingle(id)
} else {
watching, err = watcher.WatchAll()
}
if err != nil {
server.error(err, w)
return
}

// TODO: This is one watch per connection. We want to multiplex, so that
// multiple watches of the same thing don't create two watches downstream.
watchServer := &WatchServer{watching}
if req.Header.Get("Connection") == "Upgrade" && req.Header.Get("Upgrade") == "websocket" {
websocket.Handler(watchServer.HandleWS).ServeHTTP(httplog.Unlogged(w), req)
} else {
watchServer.ServeHTTP(w, req)
}
return
}

server.notFound(w, req)
}

// WatchServer serves a watch.Interface over a websocket or vanilla HTTP.
type WatchServer struct {
watching watch.Interface
}

// HandleWS implements a websocket handler.
func (w *WatchServer) HandleWS(ws *websocket.Conn) {
done := make(chan struct{})
go func() {
var unused interface{}
// Expect this to block until the connection is closed. Client should not
// send anything.
websocket.JSON.Receive(ws, &unused)
close(done)
}()
for {
select {
case <-done:
w.watching.Stop()
return
case event, ok := <-w.watching.ResultChan():
if !ok {
// End of results.
return
}
wireFormat, err := api.Encode(event.Object)
if err != nil {
glog.Errorf("error encoding %#v: %v", event.Object, err)
return
}
err = websocket.JSON.Send(ws, &api.WatchEvent{
Type: event.Type,
EmbeddedObject: wireFormat,
})
if err != nil {
// Client disconnect.
w.watching.Stop()
return
}
}
}
}

// ServeHTTP serves a series of JSON encoded events via straight HTTP with
// Transfer-Encoding: chunked.
func (self *WatchServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
loggedW := httplog.LogOf(w)
w = httplog.Unlogged(w)

cn, ok := w.(http.CloseNotifier)
if !ok {
loggedW.Addf("unable to get CloseNotifier")
http.NotFound(loggedW, req)
return
}
flusher, ok := w.(http.Flusher)
if !ok {
loggedW.Addf("unable to get Flusher")
http.NotFound(loggedW, req)
return
}

loggedW.Header().Set("Transfer-Encoding", "chunked")
loggedW.WriteHeader(http.StatusOK)
flusher.Flush()

encoder := json.NewEncoder(w)
for {
select {
case <-cn.CloseNotify():
self.watching.Stop()
return
case event, ok := <-self.watching.ResultChan():
if !ok {
// End of results.
return
}
wireFormat, err := api.Encode(event.Object)
if err != nil {
glog.Errorf("error encoding %#v: %v", event.Object, err)
return
}
err = encoder.Encode(&api.WatchEvent{
Type: event.Type,
EmbeddedObject: wireFormat,
})
if err != nil {
// Client disconnect.
self.watching.Stop()
return
}
flusher.Flush()
}
}
}
Loading

0 comments on commit eda30d4

Please sign in to comment.