summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--src/hub/src/spreadspace.org/sfive/s5srvForwardEs.go37
1 files changed, 14 insertions, 23 deletions
diff --git a/src/hub/src/spreadspace.org/sfive/s5srvForwardEs.go b/src/hub/src/spreadspace.org/sfive/s5srvForwardEs.go
index 8fb547b..b80e111 100644
--- a/src/hub/src/spreadspace.org/sfive/s5srvForwardEs.go
+++ b/src/hub/src/spreadspace.org/sfive/s5srvForwardEs.go
@@ -6,7 +6,6 @@ import (
"fmt"
"io/ioutil"
"net/http"
- "strconv"
"strings"
"time"
)
@@ -66,26 +65,15 @@ func (self StatsSinkServer) getLastUpdateEs(baseurl string, client *http.Client)
latestId = -1
return
}
- idstr := idstrcntr.(*string)
-
- tid, errl := strconv.ParseInt(*idstr, 10, 32)
- if errl != nil {
- s5l.Printf("fwd-es: invalid value: %v\n", errl)
- err = errl
- return
- }
+ tid := idstrcntr.(float64)
latestId = int(tid)
}
return
}
-type esUpdate struct {
- Data StatisticsData `json:"update"`
-}
-
func (self StatsSinkServer) handleForwardingToElasticSearch(baseurl string, client *http.Client) {
- url := baseurl + "/dataupdate/_bulk"
+ url := baseurl + "/_bulk"
tryResync:
for {
lastId, _, err := self.getLastUpdateEs(baseurl, client)
@@ -112,21 +100,23 @@ tryResync:
continue nextBatch
}
- postData := make([]byte, 4096)
+ postData := bytes.Buffer{}
for _, update := range updates {
- data, err := json.Marshal(esUpdate{Data: update})
+ data, err := json.Marshal(update)
if err != nil {
s5l.Panicf("fwd-es: encode failed: %v\n", err)
}
- postData = append(postData, ([]byte)("\n")...)
- postData = append(postData, data...)
+ postData.WriteString(`{"index":{"_type":"dataupdate"}}`)
+ postData.WriteRune('\n')
+ postData.Write(data)
+ postData.WriteRune('\n')
}
- s5tl.Printf("fwd-es: marshalled: %v", (string)(postData))
+ //s5tl.Printf("fwd-es: marshalled:\n%v\n", (string)(postData.Bytes()))
s5l.Printf("fwd-es: marshal OK")
- resp, err := client.Post(url, "text/plain", bytes.NewBuffer(postData))
+ resp, err := client.Post(url, "application/json", bytes.NewReader(postData.Bytes()))
if err != nil {
s5l.Printf("fwd-es: post failed: %v\n", err)
@@ -134,17 +124,18 @@ tryResync:
continue tryResync
}
- resp.Body.Close()
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
- s5l.Printf("fwd-es: post failed: %s\n", resp.Status)
+ body, _ := ioutil.ReadAll(resp.Body)
+ s5l.Printf("fwd-es: post failed: %s\n%s\n", resp.Status, body)
time.Sleep(1 * time.Second)
continue tryResync
}
+ resp.Body.Close()
s5l.Printf("fwd-es: all posts OK")
lastId = findMaxId(updates)
s5l.Printf("fwd-es: new lastid: %d", lastId)
- time.Sleep(1 * time.Second)
+ //time.Sleep(1 * time.Second)
}
}
}