Merge branch 'stream'

Just to get a couple of improvements and fixes. Closes oklog#30.
pbennett · Jan 24, 2017 · 1bdfb6f · 1bdfb6f
2 parents e1fef80 + 00b9845
commit 1bdfb6f
Show file tree

Hide file tree

Showing 9 changed files with 587 additions and 113 deletions.
diff --git a/cmd/oklog/ingeststore.go b/cmd/oklog/ingeststore.go
@@ -302,7 +302,8 @@ func runIngestStore(args []string) error {
 		Help:      "Number of peers in the cluster from this node's perspective.",
 	}, func() float64 { return float64(peer.ClusterSize()) }))
 
-	// Create the HTTP client we'll use to consume segments.
+	// Create the HTTP client we'll use for various purposes.
+	// TODO(pb): audit to see if we need purpose-built clients
 	httpClient := &http.Client{
 		Transport: &http.Transport{
 			Proxy: http.ProxyFromEnvironment,
@@ -422,6 +423,7 @@ func runIngestStore(args []string) error {
 			mux.Handle("/store/", http.StripPrefix("/store", store.NewAPI(
 				peer,
 				storeLog,
+				httpClient,
 				replicatedSegments.WithLabelValues("ingress"),
 				replicatedBytes.WithLabelValues("ingress"),
 				apiDuration,

diff --git a/cmd/oklog/query.go b/cmd/oklog/query.go
@@ -103,6 +103,7 @@ func runQuery(args []string) error {
 	if err != nil {
 		return err
 	}
+	verbosePrintf("GET %s\n", req.URL.String())
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
 		return err
@@ -112,14 +113,14 @@ func runQuery(args []string) error {
 	result.DecodeFrom(resp)
 
 	qtype := "normal string"
-	if result.Regex {
+	if result.Params.Regex {
 		qtype = "regular expression"
 	}
 
 	verbosePrintf("Response in %s\n", time.Since(begin))
-	verbosePrintf("Queried from %s\n", result.From)
-	verbosePrintf("Queried to %s\n", result.To)
-	verbosePrintf("Queried %s %q\n", qtype, result.Q)
+	verbosePrintf("Queried from %s\n", result.Params.From)
+	verbosePrintf("Queried to %s\n", result.Params.To)
+	verbosePrintf("Queried %s %q\n", qtype, result.Params.Q)
 	verbosePrintf("%d node(s) queried\n", result.NodesQueried)
 	verbosePrintf("%d segment(s) queried\n", result.SegmentsQueried)
 	verbosePrintf("%dB (%dMiB) maximum data set size\n", result.MaxDataSetSize, result.MaxDataSetSize/(1024*1024))

diff --git a/cmd/oklog/store.go b/cmd/oklog/store.go
@@ -198,7 +198,8 @@ func runStore(args []string) error {
 		Help:      "Number of peers in the cluster from this node's perspective.",
 	}, func() float64 { return float64(peer.ClusterSize()) }))
 
-	// Create the HTTP client we'll use to consume segments.
+	// Create the HTTP client we'll use for various purposes.
+	// TODO(pb): audit to see if we need purpose-built clients
 	httpClient := &http.Client{
 		Transport: &http.Transport{
 			Proxy: http.ProxyFromEnvironment,
@@ -268,6 +269,7 @@ func runStore(args []string) error {
 			mux.Handle("/store/", http.StripPrefix("/store", store.NewAPI(
 				peer,
 				storeLog,
+				httpClient,
 				replicatedSegments.WithLabelValues("ingress"),
 				replicatedBytes.WithLabelValues("ingress"),
 				apiDuration,