Added a basic api interface for events

Author: kateknister

api/handler.go

@@ -23,10 +23,12 @@ import (
 	"path"
 	"regexp"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 
 	"github.com/golang/glog"
+	"github.com/google/cadvisor/events"
 	"github.com/google/cadvisor/info"
 	"github.com/google/cadvisor/manager"
 )
@@ -125,6 +127,7 @@ func writeResult(res interface{}, w http.ResponseWriter) error {
 	w.Header().Set("Content-Type", "application/json")
 	w.Write(out)
 	return nil
+
 }
 
 func getContainerInfoRequest(body io.ReadCloser) (*info.ContainerInfoRequest, error) {
@@ -142,6 +145,74 @@ func getContainerInfoRequest(body io.ReadCloser) (*info.ContainerInfoRequest, er
 	return &query, nil
 }
 
+// The user can set any or none of the following arguments in any order
+// with any twice defined arguments being assigned the first value.
+// If the value type for the argument is wrong the field will be assumed to be
+// unassigned
+// bools: historical, subcontainers, oom_events, creation_events, deletion_events
+// ints: max_events, start_time (unix timestamp), end_time (unix timestamp)
+// example r.URL: http://localhost:8080/api/v1.3/events?oom_events=true&historical=true&max_events=10
+func getEventRequest(r *http.Request) (*events.Request, bool, error) {
+	query := events.NewRequest()
+	getHistoricalEvents := false
+
+	urlMap := r.URL.Query()
+
+	if val, ok := urlMap["historical"]; ok {
+		newBool, err := strconv.ParseBool(val[0])
+		if err == nil {
+			getHistoricalEvents = newBool
+		}
+	}
+	if val, ok := urlMap["subcontainers"]; ok {
+		newBool, err := strconv.ParseBool(val[0])
+		if err == nil {
+			query.IncludeSubcontainers = newBool
+		}
+	}
+	if val, ok := urlMap["oom_events"]; ok {
+		newBool, err := strconv.ParseBool(val[0])
+		if err == nil {
+			query.EventType[events.TypeOom] = newBool
+		}
+	}
+	if val, ok := urlMap["creation_events"]; ok {
+		newBool, err := strconv.ParseBool(val[0])
+		if err == nil {
+			query.EventType[events.TypeContainerCreation] = newBool
+		}
+	}
+	if val, ok := urlMap["deletion_events"]; ok {
+		newBool, err := strconv.ParseBool(val[0])
+		if err == nil {
+			query.EventType[events.TypeContainerDeletion] = newBool
+		}
+	}
+	if val, ok := urlMap["max_events"]; ok {
+		newInt, err := strconv.Atoi(val[0])
+		if err == nil {
+			query.MaxEventsReturned = int(newInt)
+		}
+	}
+	if val, ok := urlMap["start_time"]; ok {
+		newTime, err := time.Parse(time.RFC3339, val[0])
+		if err == nil {
+			query.StartTime = newTime
+		}
+	}
+	if val, ok := urlMap["end_time"]; ok {
+		newTime, err := time.Parse(time.RFC3339, val[0])
+		if err == nil {
+			query.EndTime = newTime
+		}
+	}
+
+	glog.V(2).Infof(
+		"%v was returned in api/handler.go:getEventRequest from the url rawQuery %v",
+		query, r.URL.RawQuery)
+	return query, getHistoricalEvents, nil
+}
+
 func getContainerName(request []string) string {
 	return path.Join("/", strings.Join(request, "/"))
 }

api/versions.go

@@ -19,6 +19,7 @@ import (
 	"net/http"
 
 	"github.com/golang/glog"
+	"github.com/google/cadvisor/events"
 	"github.com/google/cadvisor/info"
 	"github.com/google/cadvisor/manager"
 )
@@ -30,6 +31,7 @@ const (
 	dockerApi        = "docker"
 	summaryApi       = "summary"
 	specApi          = "spec"
+	eventsApi        = "events"
 )
 
 // Interface for a cAdvisor API version
@@ -49,8 +51,11 @@ func getApiVersions() []ApiVersion {
 	v1_0 := &version1_0{}
 	v1_1 := newVersion1_1(v1_0)
 	v1_2 := newVersion1_2(v1_1)
-	v2_0 := newVersion2_0(v1_2)
-	return []ApiVersion{v1_0, v1_1, v1_2, v2_0}
+	v1_3 := newVersion1_3(v1_2)
+	v2_0 := newVersion2_0(v1_3)
+
+	return []ApiVersion{v1_0, v1_1, v1_2, v1_3, v2_0}
+
 }
 
 // API v1.0
@@ -227,12 +232,72 @@ func (self *version1_2) HandleRequest(requestType string, request []string, m ma
 	}
 }
 
-// v2.0 builds on v1.2
-type version2_0 struct {
+// API v1.3
+
+type version1_3 struct {
 	baseVersion *version1_2
 }
 
-func newVersion2_0(v *version1_2) *version2_0 {
+// v1.3 builds on v1.2.
+func newVersion1_3(v *version1_2) *version1_3 {
+	return &version1_3{
+		baseVersion: v,
+	}
+}
+
+func (self *version1_3) Version() string {
+	return "v1.3"
+}
+
+func (self *version1_3) SupportedRequestTypes() []string {
+	return append(self.baseVersion.SupportedRequestTypes(), eventsApi)
+}
+
+func (self *version1_3) HandleRequest(requestType string, request []string, m manager.Manager, w http.ResponseWriter, r *http.Request) error {
+	switch requestType {
+	case eventsApi:
+		query, eventsFromAllTime, err := getEventRequest(r)
+		if err != nil {
+			return err
+		}
+		glog.V(2).Infof("Api - Events(%v)", query)
+
+		if eventsFromAllTime {
+			allEvents, err := m.GetPastEvents(query)
+			if err != nil {
+				return err
+			}
+			return writeResult(allEvents, w)
+		} else {
+			// every time URL is entered to watch, a channel is created here
+			eventChannel := make(chan *events.Event, 10)
+			err = m.WatchForEvents(query, eventChannel)
+
+			defer close(eventChannel)
+			currentEventSet := make(events.EventSlice, 0)
+			for ev := range eventChannel {
+				// todo: implement write-as-received writeResult method
+				currentEventSet = append(currentEventSet, ev)
+				err = writeResult(currentEventSet, w)
+				if err != nil {
+					return err
+				}
+			}
+		}
+		return nil
+	default:
+		return self.baseVersion.HandleRequest(requestType, request, m, w, r)
+	}
+}
+
+// API v2.0
+
+// v2.0 builds on v1.3
+type version2_0 struct {
+	baseVersion *version1_3
+}
+
+func newVersion2_0(v *version1_3) *version2_0 {
 	return &version2_0{
 		baseVersion: v,
 	}

api/versions_test.go

@@ -0,0 +1,94 @@
+// Copyright 2015 Google Inc. 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 api
+
+import (
+	"io"
+	"net/http"
+	"reflect"
+	"testing"
+
+	"github.com/google/cadvisor/events"
+	"github.com/stretchr/testify/assert"
+)
+
+// returns an http.Request pointer for an input url test string
+func makeHTTPRequest(requestURL string, t *testing.T) *http.Request {
+	dummyReader, _ := io.Pipe()
+	r, err := http.NewRequest("GET", requestURL, dummyReader)
+	assert.Nil(t, err)
+	return r
+}
+
+func TestGetEventRequestBasicRequest(t *testing.T) {
+	r := makeHTTPRequest("http://localhost:8080/api/v1.3/events?oom_events=true&historical=true&max_events=10", t)
+	expectedQuery := &events.Request{
+		EventType: map[events.EventType]bool{
+			events.TypeOom: true,
+		},
+		MaxEventsReturned: 10,
+	}
+
+	receivedQuery, getHistoricalEvents, err := getEventRequest(r)
+
+	if !reflect.DeepEqual(expectedQuery, receivedQuery) {
+		t.Errorf("expected %v but received %v", expectedQuery, receivedQuery)
+	}
+	assert.True(t, getHistoricalEvents)
+	assert.Nil(t, err)
+}
+
+func TestGetEventEmptyRequest(t *testing.T) {
+	r := makeHTTPRequest("", t)
+	expectedQuery := events.NewRequest()
+
+	receivedQuery, getHistoricalEvents, err := getEventRequest(r)
+
+	if !reflect.DeepEqual(expectedQuery, receivedQuery) {
+		t.Errorf("expected %v but received %v", expectedQuery, receivedQuery)
+	}
+	assert.False(t, getHistoricalEvents)
+	assert.Nil(t, err)
+}
+
+func TestGetEventEmptyStringRequest(t *testing.T) {
+	r := makeHTTPRequest("/", t)
+	expectedQuery := events.NewRequest()
+
+	receivedQuery, getHistoricalEvents, err := getEventRequest(r)
+
+	if !reflect.DeepEqual(expectedQuery, receivedQuery) {
+		t.Errorf("expected %v but received %v", expectedQuery, receivedQuery)
+	}
+	assert.False(t, getHistoricalEvents)
+	assert.Nil(t, err)
+}
+
+func TestGetEventRequestDoubleArgument(t *testing.T) {
+	r := makeHTTPRequest("http://localhost:8080/api/v1.3/events?historical=true&oom_events=true&oom_events=false", t)
+	expectedQuery := &events.Request{
+		EventType: map[events.EventType]bool{
+			events.TypeOom: true,
+		},
+	}
+
+	receivedQuery, getHistoricalEvents, err := getEventRequest(r)
+
+	if !reflect.DeepEqual(expectedQuery, receivedQuery) {
+		t.Errorf("expected %v but received %v", expectedQuery, receivedQuery)
+	}
+	assert.True(t, getHistoricalEvents)
+	assert.Nil(t, err)
+}

events/handler.go

@@ -103,7 +103,7 @@ type Request struct {
 	// allows the caller to put a limit on how many
 	// events they receive. If there are more events than MaxEventsReturned
 	// then the most chronologically recent events in the time period
-	// specified are returned
+	// specified are returned. Must be >= 1
 	MaxEventsReturned int
 	// the absolute container name for which the event occurred
 	ContainerName string

manager/manager.go

@@ -74,6 +74,9 @@ type Manager interface {
 
 	// Get events streamed through passedChannel that fit the request.
 	WatchForEvents(request *events.Request, passedChannel chan *events.Event) error
+
+	// Get past events that have been detected and that fit the request.
+	GetPastEvents(request *events.Request) (events.EventSlice, error)
 }
 
 // New takes a memory storage and returns a new manager.
@@ -474,6 +477,19 @@ func (m *manager) createContainer(containerName string) error {
 
 	// Start the container's housekeeping.
 	cont.Start()
+
+	newEvent := &events.Event{
+		EventType:     events.TypeContainerCreation,
+		Timestamp:     time.Now(),
+		ContainerName: containerName,
+		EventData:     cont,
+	}
+	err = m.eventHandler.AddEvent(newEvent)
+	if err != nil {
+		glog.Errorf("Failed to add event %v, got error: %v", newEvent, err)
+		return err
+	}
+
 	return nil
 }
 
@@ -505,6 +521,18 @@ func (m *manager) destroyContainer(containerName string) error {
 		})
 	}
 	glog.Infof("Destroyed container: %q (aliases: %v, namespace: %q)", containerName, cont.info.Aliases, cont.info.Namespace)
+
+	newEvent := &events.Event{
+		EventType:     events.TypeContainerDeletion,
+		Timestamp:     time.Now(),
+		ContainerName: containerName,
+		EventData:     cont,
+	}
+	err = m.eventHandler.AddEvent(newEvent)
+	if err != nil {
+		glog.Errorf("Failed to add event %v, got error: %v", newEvent, err)
+		return err
+	}
 	return nil
 }
 
@@ -596,8 +624,8 @@ func (self *manager) watchForNewContainers(quit chan error) error {
 	}
 
 	// Register for new subcontainers.
-	events := make(chan container.SubcontainerEvent, 16)
-	err := root.handler.WatchSubcontainers(events)
+	eventsChannel := make(chan container.SubcontainerEvent, 16)
+	err := root.handler.WatchSubcontainers(eventsChannel)
 	if err != nil {
 		return err
 	}
@@ -612,15 +640,16 @@ func (self *manager) watchForNewContainers(quit chan error) error {
 	go func() {
 		for {
 			select {
-			case event := <-events:
+			case event := <-eventsChannel:
 				switch {
 				case event.EventType == container.SubcontainerAdd:
 					err = self.createContainer(event.Name)
 				case event.EventType == container.SubcontainerDelete:
 					err = self.destroyContainer(event.Name)
-				}
-				if err != nil {
-					glog.Warning("Failed to process watch event: %v", err)
+					if err != nil {
+						glog.Warning("Failed to process watch event: %v", err)
+					}
+
 				}
 			case <-quit:
 				// Stop processing events if asked to quit.
@@ -668,3 +697,8 @@ func (self *manager) watchForNewOoms() error {
 func (self *manager) WatchForEvents(request *events.Request, passedChannel chan *events.Event) error {
 	return self.eventHandler.WatchEvents(passedChannel, request)
 }
+
+// can be called by the api which will return all events satisfying the request
+func (self *manager) GetPastEvents(request *events.Request) (events.EventSlice, error) {
+	return self.eventHandler.GetEvents(request)
+}