Merge pull request #26 from candysmurf/big-chunck

SDI-2279: Plugin doesn't handle publishing big chunk

glide.yaml

@@ -0,0 +1,44 @@
+package: github.com/intelsdi-x/snap-plugin-publisher-opentsdb
+import:
+- package: github.com/Sirupsen/logrus
+  version: cd7d1bbe41066b6c1f19780f895901052150a575
+- package: github.com/gopherjs/gopherjs
+  version: 4b53e1bddba0e2f734514aeb6c02db652f4c6fe8
+  subpackages:
+  - js
+- package: github.com/intelsdi-x/snap
+  version: 005961655e59e33a1515a432b1c6e6074d152519
+  subpackages:
+  - control/plugin
+  - control/plugin/cpolicy
+  - control/plugin/encoding
+  - control/plugin/encrypter
+  - core
+  - core/cdata
+  - core/ctypes
+  - core/serror
+  - pkg/ctree
+  - pkg/schedule
+  - pkg/stringutils
+  - scheduler/wmap
+- package: github.com/jtolds/gls
+  version: 8ddce2a84170772b95dd5d576c48d517b22cac63
+- package: github.com/robfig/cron
+  version: 32d9c273155a0506d27cf73dd1246e86a470997e
+- package: github.com/smartystreets/assertions
+  version: 443d812296a84445c202c085f19e18fc238f8250
+  subpackages:
+  - internal/go-render/render
+  - internal/oglematchers
+- package: github.com/smartystreets/goconvey
+  version: 995f5b2e021c69b8b028ba6d0b05c1dd500783db
+  subpackages:
+  - convey
+  - convey/gotest
+  - convey/reporting
+- package: golang.org/x/sys
+  version: 7f918dd405547ecb864d14a8ecbbfe205b5f930f
+  subpackages:
+  - unix
+- package: gopkg.in/yaml.v2
+  version: c1cd2254a6dd314c9d73c338c12688c9325d85c6

opentsdb/httpclient.go

@@ -32,8 +32,9 @@ import (
 
 const (
 	putEndPoint     = "/api/put"
-	contentTypeJson = "application/json"
+	contentTypeJSON = "application/json"
 	userAgent       = "snap-publisher"
+	maxChunkLength  = 25
 )
 
 type HttpClient struct {
@@ -58,7 +59,7 @@ func NewClient(url string, timeout time.Duration) *HttpClient {
 	}
 }
 
-func (hc *HttpClient) getUrl() string {
+func (hc *HttpClient) getURL() string {
 	u := url.URL{
 		Scheme: "http",
 		Host:   hc.url,
@@ -67,16 +68,40 @@ func (hc *HttpClient) getUrl() string {
 	return u.String()
 }
 
-// Post stores slides of Datapoint to OpenTSDB
-func (hc *HttpClient) Post(dps []DataPoint) error {
-	url := hc.getUrl()
+// Save saves data points in maxChunkLength size.
+func (hc *HttpClient) Save(dps []DataPoint) error {
+	url := hc.getURL()
+
+	loop := len(dps) / maxChunkLength
+	start := 0
+	end := start
+	for i := 0; i < loop; i++ {
+		end += maxChunkLength
+		chunk := dps[start:end]
+		start = end
+		err := hc.post(url, chunk)
+		if err != nil {
+			return err
+		}
+	}
+
+	remainder := len(dps) % maxChunkLength
+	if remainder > 0 {
+		end = start + remainder
+		chunk := dps[start:end]
+		return hc.post(url, chunk)
+	}
+	return nil
+}
 
+// post stores a slice of Datapoint to OpenTSDB
+func (hc *HttpClient) post(url string, dps []DataPoint) error {
 	buf, err := json.Marshal(dps)
 	if err != nil {
 		return err
 	}
 
-	resp, err := hc.httpClient.Post(url, contentTypeJson, bytes.NewReader(buf))
+	resp, err := hc.httpClient.Post(url, contentTypeJSON, bytes.NewReader(buf))
 	if err != nil {
 		return err
 	}

opentsdb/opentsdb.go

@@ -38,7 +38,7 @@ import (
 
 const (
 	name       = "opentsdb"
-	version    = 8
+	version    = 9
 	pluginType = plugin.PublisherPluginType
 	timeout    = 5
 	host       = "host"
@@ -154,7 +154,7 @@ func (p *opentsdbPublisher) Publish(contentType string, content []byte, config m
 
 	td := time.Duration(timeout * time.Second)
 	con := NewClient(u.String(), td)
-	err = con.Post(pts)
+	err = con.Save(pts)
 	if err != nil {
 		logger.Printf("Error: '%s' posting metrics: %+v", err.Error(), metrics)
 		return err