Merge pull request #1762 from honeycombio/honeycomb/pr

Add Honeycomb Sink for Heaptser

common/honeycomb/dummy_honeycomb.go

@@ -0,0 +1,40 @@
+// Copyright 2017 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 honeycomb
+
+type BatchPointsSavedToHoneycomb struct {
+	BatchPoint *BatchPoint
+}
+
+type FakeHoneycombClient struct {
+	BatchPoints []*BatchPoint
+}
+
+func NewFakeHoneycombClient() *FakeHoneycombClient {
+	return &FakeHoneycombClient{[]*BatchPoint{}}
+}
+
+func (client *FakeHoneycombClient) SendBatch(batch Batch) error {
+	for _, batchpoint := range batch {
+		client.BatchPoints = append(client.BatchPoints, batchpoint)
+	}
+	return nil
+}
+
+var Config = config{
+	Dataset:  "fake",
+	WriteKey: "fakekey",
+	APIHost:  "fakehost",
+}

common/honeycomb/honeycomb.go

@@ -0,0 +1,125 @@
+// Copyright 2017 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 honeycomb
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"os"
+	"path"
+	"time"
+
+	"github.com/golang/glog"
+)
+
+type config struct {
+	APIHost  string
+	Dataset  string
+	WriteKey string
+}
+
+func BuildConfig(uri *url.URL) (*config, error) {
+	opts := uri.Query()
+
+	config := &config{
+		WriteKey: os.Getenv("HONEYCOMB_WRITEKEY"),
+		APIHost:  "https://api.honeycomb.io/",
+		Dataset:  "heapster",
+	}
+
+	if len(opts["writekey"]) >= 1 {
+		config.WriteKey = opts["writekey"][0]
+	}
+
+	if len(opts["apihost"]) >= 1 {
+		config.APIHost = opts["apihost"][0]
+	}
+
+	if len(opts["dataset"]) >= 1 {
+		config.Dataset = opts["dataset"][0]
+	}
+
+	if config.WriteKey == "" {
+		return nil, errors.New("Failed to find honeycomb API write key")
+	}
+
+	return config, nil
+}
+
+type Client interface {
+	SendBatch(batch Batch) error
+}
+
+type HoneycombClient struct {
+	config     config
+	httpClient http.Client
+}
+
+func NewClient(uri *url.URL) (*HoneycombClient, error) {
+	config, err := BuildConfig(uri)
+	if err != nil {
+		return nil, err
+	}
+	return &HoneycombClient{config: *config}, nil
+}
+
+type BatchPoint struct {
+	Data      interface{}
+	Timestamp time.Time
+}
+
+type Batch []*BatchPoint
+
+func (c *HoneycombClient) SendBatch(batch Batch) error {
+	if len(batch) == 0 {
+		// Nothing to send
+		return nil
+	}
+	buf := new(bytes.Buffer)
+	err := json.NewEncoder(buf).Encode(batch)
+	if err != nil {
+		return err
+	}
+	err = c.makeRequest(buf)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (c *HoneycombClient) makeRequest(body io.Reader) error {
+	url, err := url.Parse(c.config.APIHost)
+	if err != nil {
+		return err
+	}
+	url.Path = path.Join(url.Path, "/1/batch", c.config.Dataset)
+	req, err := http.NewRequest("POST", url.String(), body)
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Add("X-Honeycomb-Team", c.config.WriteKey)
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		glog.Warningf("Failed to send event: %v", err)
+		return err
+	}
+	defer resp.Body.Close()
+	ioutil.ReadAll(resp.Body)
+	return nil
+}

common/honeycomb/honeycomb_test.go

@@ -0,0 +1,60 @@
+// Copyright 2017 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 honeycomb
+
+import (
+	"net/http/httptest"
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	util "k8s.io/client-go/util/testing"
+)
+
+func TestHoneycombClientWrite(t *testing.T) {
+	handler := util.FakeHandler{
+		StatusCode:   202,
+		ResponseBody: "",
+		T:            t,
+	}
+	server := httptest.NewServer(&handler)
+	defer server.Close()
+
+	stubURL, err := url.Parse("?writekey=testkey&dataset=testdataset&apihost=" + server.URL)
+
+	assert.NoError(t, err)
+
+	config, err := BuildConfig(stubURL)
+
+	assert.Equal(t, config.WriteKey, "testkey")
+	assert.Equal(t, config.APIHost, server.URL)
+	assert.Equal(t, config.Dataset, "testdataset")
+
+	assert.NoError(t, err)
+
+	client, _ := NewClient(stubURL)
+
+	err = client.SendBatch([]*BatchPoint{
+		{
+			Data:      "test",
+			Timestamp: time.Now(),
+		},
+	})
+
+	assert.NoError(t, err)
+
+	handler.ValidateRequestCount(t, 1)
+}

docs/sink-configuration.md

@@ -289,6 +289,24 @@ For example,
 
 The librato sink currently only works with accounts, which support [tagged metrics](https://www.librato.com/docs/kb/faq/account_questions/tags_or_sources/).
 
+### Honeycomb
+
+This sink supports both monitoring metrics and events.
+
+To use the Honeycomb sink add the following flag:
+
+    --sink="honeycomb:<?<OPTIONS>>"
+
+Options can be set in query string, like this:
+
+* `dataset` - Honeycomb Dataset to which to publish metrics/events
+* `writekey` - Honeycomb Write Key for your account
+* `apihost` - Option to send metrics to a different host (default: https://api.honeycomb.com) (optional)
+
+For example,
+
+    --sink="honeycomb:?dataset=mydataset&writekey=secretwritekey"
+
 ## Using multiple sinks
 
 Heapster can be configured to send k8s metrics and events to multiple sinks by specifying the`--sink=...` flag multiple times.

docs/sink-owners.md

@@ -37,3 +37,4 @@ List of Owners
 | Graphite        | :heavy_check_mark: | :x:                | @jsoriano / @theairkit                        | :new: #1341    |
 | Wavefront       | :heavy_check_mark: | :x:                | @ezeev                                        | :ok:           |
 | Librato         | :heavy_check_mark: | :x:                | @johanneswuerbach                             | :ok:           |
+| Honeycomb       | :heavy_check_mark: | :heavy_check_mark: | @emfree                                       | :new: #1762    |

events/sinks/factory.go

@@ -21,6 +21,7 @@ import (
 	"k8s.io/heapster/events/core"
 	"k8s.io/heapster/events/sinks/elasticsearch"
 	"k8s.io/heapster/events/sinks/gcl"
+	"k8s.io/heapster/events/sinks/honeycomb"
 	"k8s.io/heapster/events/sinks/influxdb"
 	"k8s.io/heapster/events/sinks/kafka"
 	"k8s.io/heapster/events/sinks/log"
@@ -46,6 +47,8 @@ func (this *SinkFactory) Build(uri flags.Uri) (core.EventSink, error) {
 		return kafka.NewKafkaSink(&uri.Val)
 	case "riemann":
 		return riemann.CreateRiemannSink(&uri.Val)
+	case "honeycomb":
+		return honeycomb.NewHoneycombSink(&uri.Val)
 	default:
 		return nil, fmt.Errorf("Sink not recognized: %s", uri.Key)
 	}

events/sinks/honeycomb/driver.go

@@ -0,0 +1,95 @@
+// Copyright 2017 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 honeycomb
+
+import (
+	"net/url"
+	"sync"
+
+	"github.com/golang/glog"
+	kube_api "k8s.io/client-go/pkg/api/v1"
+	honeycomb_common "k8s.io/heapster/common/honeycomb"
+	event_core "k8s.io/heapster/events/core"
+)
+
+type honeycombSink struct {
+	client honeycomb_common.Client
+	sync.Mutex
+}
+
+type exportedData struct {
+	Namespace       string `json:"namespace"`
+	Kind            string `json:"kind"`
+	Name            string `json:"name"`
+	SubObject       string `json:"subobject"`
+	SourceComponent string `json:"source.component"`
+	SourceHost      string `json:"source.host"`
+	Count           int32  `json:"count"`
+	Type            string `json:"type"`
+	Reason          string `json:"reason"`
+	Message         string `json:"message"`
+}
+
+func getExportedData(e *kube_api.Event) *exportedData {
+	return &exportedData{
+		Namespace:       e.InvolvedObject.Namespace,
+		Kind:            e.InvolvedObject.Kind,
+		Name:            e.InvolvedObject.Name,
+		SubObject:       e.InvolvedObject.FieldPath,
+		SourceComponent: e.Source.Component,
+		SourceHost:      e.Source.Host,
+		Count:           e.Count,
+		Reason:          e.Reason,
+		Type:            e.Type,
+		Message:         e.Message,
+	}
+}
+
+func (sink *honeycombSink) ExportEvents(eventBatch *event_core.EventBatch) {
+	sink.Lock()
+	defer sink.Unlock()
+	exportedBatch := make(honeycomb_common.Batch, len(eventBatch.Events))
+	for i, event := range eventBatch.Events {
+		data := getExportedData(event)
+		exportedBatch[i] = &honeycomb_common.BatchPoint{
+			Data:      data,
+			Timestamp: event.LastTimestamp.UTC(),
+		}
+	}
+	err := sink.client.SendBatch(exportedBatch)
+	if err != nil {
+		glog.Warningf("Failed to send event: %v", err)
+		return
+	}
+}
+
+func (sink *honeycombSink) Stop() {}
+
+func (sink *honeycombSink) Name() string {
+	return "Honeycomb Sink"
+}
+
+func NewHoneycombSink(uri *url.URL) (event_core.EventSink, error) {
+	client, err := honeycomb_common.NewClient(uri)
+	if err != nil {
+		return nil, err
+	}
+	sink := &honeycombSink{
+		client: client,
+	}
+
+	return sink, nil
+
+}