Merge branch 'master' into fix/update-docker-image

.circleci/config.yml

@@ -31,10 +31,10 @@ jobs:
           command: go build
       - run:
           name: Run Tests
-          command: go test -race -cover -coverprofile=${CIRCLE_COVERAGE_REPORTS}/coverage.out ./... | go-junit-report > ${CIRCLE_TEST_REPORTS}/junit.xml
+          command: go test -race -cover -coverprofile=${CIRCLE_COVERAGE_REPORT}/coverage.out ./... | go-junit-report > ${CIRCLE_TEST_REPORTS}/junit.xml
       - run:
           name: Upload Coverage
-          command: goveralls -coverprofile=${CIRCLE_COVERAGE_REPORTS}/coverage.out -service=circle-ci -repotoken=${COVERALLS_TOKEN}
+          command: goveralls -coverprofile=${CIRCLE_COVERAGE_REPORT}/coverage.out -service=circle-ci -repotoken=${COVERALLS_TOKEN}
       - store_test_results:
           path: /tmp/test-results
 
@@ -55,4 +55,4 @@ workflows:
       - build
       - dockerfile:
           requires:
-            - build
+            - build

Dockerfile

@@ -28,4 +28,4 @@ WORKDIR /
 COPY --from=0 /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
 COPY --from=0 /artifacts/* /
 
-CMD [ "/content-collection-unfolder" ]
+CMD [ "/content-collection-unfolder" ]

config.go

@@ -19,6 +19,7 @@ type serviceConfig struct {
 	kafkaAddr                  *string
 	kafkaHostname              *string
 	kafkaAuth                  *string
+	requestTimeout             *int
 }
 
 func createServiceConfiguration(app *cli.Cli) *serviceConfig {
@@ -119,6 +120,13 @@ func createServiceConfiguration(app *cli.Cli) *serviceConfig {
 		EnvVar: "Q_AUTHORIZATION",
 	})
 
+	requestTimeout := app.Int(cli.IntOpt{
+		Name:   "request_timeout",
+		Value:  2,
+		Desc:   "timeout per request for taking contents from document store",
+		EnvVar: "REQUEST_TIMEOUT",
+	})
+
 	return &serviceConfig{
 		appSystemCode:              appSystemCode,
 		appName:                    appName,
@@ -134,6 +142,7 @@ func createServiceConfiguration(app *cli.Cli) *serviceConfig {
 		kafkaAddr:                  kafkaAddr,
 		kafkaHostname:              kafkaHostname,
 		kafkaAuth:                  kafkaAuth,
+		requestTimeout:             requestTimeout,
 	}
 }
 
@@ -153,5 +162,6 @@ func (sc *serviceConfig) toMap() map[string]interface{} {
 		"kafkaAddr":                  *sc.kafkaAddr,
 		"kafkaHostname":              *sc.kafkaHostname,
 		"kafkaAuth":                  *sc.kafkaAuth,
+		"requestTimeout":             *sc.requestTimeout,
 	}
 }

config_test.go

@@ -3,10 +3,15 @@ package main
 import (
 	"testing"
 
-	"github.com/jawher/mow.cli"
+	cli "github.com/jawher/mow.cli"
 	"github.com/stretchr/testify/assert"
 )
 
+const (
+	emptyString           = ""
+	requestTimeoutSeconds = 2
+)
+
 func TestDefaults(t *testing.T) {
 	app := cli.App("content-collection-unfolder", serviceDescription)
 
@@ -27,8 +32,8 @@ func TestDefaults(t *testing.T) {
 		assert.NotEmpty(t, configMap["writeTopic"])
 		assert.NotEmpty(t, configMap["kafkaAddr"])
 		assert.NotEmpty(t, configMap["kafkaHostname"])
-		assert.Equal(t, "", configMap["kafkaAuth"])
-
+		assert.Equal(t, emptyString, configMap["kafkaAuth"])
+		assert.Equal(t, requestTimeoutSeconds, configMap["requestTimeout"])
 	}
 
 	app.Run([]string{"content-collection-unfolder"})

helm/content-collection-unfolder/app-configs/content-collection-unfolder_delivery.yaml

@@ -11,3 +11,5 @@ env:
   RELATIONS_RESOLVER_URI: http://relations-api:8080/contentcollection/{uuid}/relations
   RELATIONS_RESOLVER_HEALTH_URI: http://relations-api:8080/__health
   Q_WRITE_TOPIC: PostPublicationEvents
+  REQUEST_TIMEOUT: '"2"'
+

helm/content-collection-unfolder/templates/deployment.yaml

@@ -54,6 +54,8 @@ spec:
             configMapKeyRef: 
               name: global-config 
               key: kafka.proxy.url.with.protocol
+        - name: REQUEST_TIMEOUT
+          value: {{ .Values.env.REQUEST_TIMEOUT }}
         ports: 
         - containerPort: 8080 
         livenessProbe: 

main.go

@@ -38,7 +38,7 @@ func main() {
 			relations.NewDefaultRelationsResolver(client, *sc.relationsResolverURI),
 			differ.NewDefaultCollectionsDiffer(),
 			fw.NewForwarder(client, *sc.writerURI),
-			res.NewContentResolver(client, *sc.contentResolverURI),
+			res.NewContentResolver(client, *sc.contentResolverURI, time.Duration(*sc.requestTimeout)*time.Second),
 			prod.NewContentProducer(producer),
 			*sc.unfoldingWhitelist,
 		)

main_test.go

@@ -29,6 +29,7 @@ const (
 	relationsResolverPath       = "/contentcollection/{uuid}/relations"
 	relationsResolverHealthPath = "/__health"
 	tid                         = "tid_test123456"
+	requestTimeoutInt           = 2e9
 )
 
 func TestAllHealthChecksBad(t *testing.T) {
@@ -273,7 +274,7 @@ func startRouting(
 			relations.NewDefaultRelationsResolver(client, relationsResolverServer.URL+relationsResolverPath),
 			differ.NewDefaultCollectionsDiffer(),
 			fw.NewForwarder(client, writerServer.URL+strings.Split(writerPath, "/{")[0]),
-			res.NewContentResolver(client, contentResolverServer.URL+contentResolverPath),
+			res.NewContentResolver(client, contentResolverServer.URL+contentResolverPath, requestTimeoutInt),
 			prod.NewContentProducer(messageProducer),
 			[]string{whitelistedCollection},
 		),

resolver/contentResolver.go

@@ -5,21 +5,25 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net/http"
+	"time"
 
 	"github.com/Financial-Times/transactionid-utils-go"
 )
 
 type ContentResolver interface {
 	ResolveContents(diffUuids []string, tid string) ([]map[string]interface{}, error)
+	ResolveContentsNew(diffUuids []string, tid string, requestTimeout time.Duration) ([]map[string]interface{}, error)
+	GetRequestTimeout() time.Duration
 }
 
 type defaultContentResolver struct {
 	contentResolverAppURI string
+	requestTimeout        time.Duration
 	httpClient            *http.Client
 }
 
-func NewContentResolver(client *http.Client, contentResolverAppURI string) ContentResolver {
-	return &defaultContentResolver{contentResolverAppURI: contentResolverAppURI, httpClient: client}
+func NewContentResolver(client *http.Client, contentResolverAppURI string, requestTimeoutArg time.Duration) ContentResolver {
+	return &defaultContentResolver{contentResolverAppURI: contentResolverAppURI, requestTimeout: requestTimeoutArg, httpClient: client}
 }
 
 func (cr *defaultContentResolver) ResolveContents(diffUuids []string, tid string) ([]map[string]interface{}, error) {
@@ -70,3 +74,78 @@ func (cr *defaultContentResolver) callContentResolverApp(diffUuids []string, tid
 
 	return resp, nil
 }
+
+func (cr *defaultContentResolver) GetRequestTimeout() time.Duration {
+	return cr.requestTimeout
+}
+
+func (cr *defaultContentResolver) ResolveContentsNew(diffUuids []string, tid string, requestTimeout time.Duration) ([]map[string]interface{}, error) {
+	return cr.callContentResolverAppNew(diffUuids, tid, requestTimeout)
+}
+
+func (cr *defaultContentResolver) callContentResolverAppNew(diffUuids []string, tid string, requestTimeout time.Duration) ([]map[string]interface{}, error) {
+	jsonResponses := make(chan []map[string]interface{}, 1)
+
+	req, err := cr.createRequest(tid)
+	if err != nil {
+		return nil, fmt.Errorf("Error calling on url [%v] for content, error was: [%v]", cr.contentResolverAppURI, err.Error())
+	}
+	httpQuery := req.URL.Query()
+	for _, diffUuid := range diffUuids {
+		httpQuery.Add("uuid", diffUuid)
+		time.Sleep(requestTimeout)
+	}
+
+	req.URL.RawQuery = httpQuery.Encode()
+	resp, err := cr.httpClient.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("Error doing request to uri=[%v], transaction_id=[%v].", cr.contentResolverAppURI, tid)
+	}
+
+	bodyAsBytes, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("Could not read response after calling [%v], transaction_id=[%v], error was: [%v]", cr.contentResolverAppURI, tid, err.Error())
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("Call to [%v] for transaction_id=[%v], responded with error statusCode [%d], error was: [%v]", cr.contentResolverAppURI, tid, resp.StatusCode, string(bodyAsBytes))
+	}
+	defer resp.Body.Close()
+	var content []map[string]interface{}
+	err = json.Unmarshal(bodyAsBytes, &content)
+	if err != nil {
+		return nil, fmt.Errorf("Could not read response body from call to [%v], transaction_id=[%v], error was: [%v]", cr.contentResolverAppURI, tid, err.Error())
+	}
+	jsonResponses <- content
+
+	return getFromResponsesChannel(jsonResponses)
+}
+
+func (cr *defaultContentResolver) createRequest(tid string) (*http.Request, error) {
+	req, err := http.NewRequest(http.MethodGet, cr.contentResolverAppURI, nil)
+	if err != nil {
+		return nil, fmt.Errorf("Error creating request to uri=[%v], transaction_id=[%v].", cr.contentResolverAppURI, tid)
+	}
+
+	req.Header.Set(transactionidutils.TransactionIDHeader, tid)
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("User-Agent", "UPP content-collection-unfolder")
+	return req, nil
+}
+
+func getFromResponsesChannel(jsonResponses chan []map[string]interface{}) ([]map[string]interface{}, error) {
+	responsesCount := len(jsonResponses)
+	var jsonResponsesArr []map[string]interface{}
+
+	for i := 0; i < responsesCount; i++ {
+		jsonResponsesArr = <-jsonResponses
+	}
+
+	close(jsonResponses)
+
+	if len(jsonResponsesArr) != 0 {
+		return jsonResponsesArr, nil
+	} else {
+		return nil, nil
+	}
+}

resolver/contentResolver_test.go

@@ -7,6 +7,7 @@ import (
 	"net/http/httptest"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
@@ -16,8 +17,8 @@ import (
 const (
 	statusWorking    = "working"
 	statusNotWorking = "notWorking"
-
-	tid = "tid_qkeqptjwji"
+	tid              = "tid_qkeqptjwji"
+	requestTimeout   = time.Second * 2
 )
 
 var contentResolver ContentResolver
@@ -50,14 +51,14 @@ func mockDSAPIBytes(appStatus string, output []byte) {
 	router.Path("/content").Handler(handlers.MethodHandler{"GET": http.HandlerFunc(contentResolverEndpointHandler)})
 	dsAPIMock = httptest.NewServer(router)
 
-	contentResolver = NewContentResolver(http.DefaultClient, dsAPIMock.URL+"/content")
+	contentResolver = NewContentResolver(http.DefaultClient, dsAPIMock.URL+"/content", requestTimeout)
 }
 
 func Test_callContentResolverApp_1_Content(t *testing.T) {
 	mockDSAPI(t, statusWorking, "document-store-api-1-content-output.json")
 
 	diffUuids := []string{"ab43b1a6-1f47-11e7-b7d3-163f5a7f229c"}
-	contents, err := contentResolver.ResolveContents(diffUuids, tid)
+	contents, err := contentResolver.ResolveContentsNew(diffUuids, tid, requestTimeout)
 	if err != nil {
 		assert.FailNow(t, "Failed retrieving contents.", err.Error())
 	}
@@ -69,19 +70,31 @@ func Test_callContentResolverApp_2_Content(t *testing.T) {
 	mockDSAPI(t, statusWorking, "document-store-api-2-content-output.json")
 
 	diffUuids := []string{"ab43b1a6-1f47-11e7-b7d3-163f5a7f229c", "70c800d8-b3e3-11e6-ba85-95d1533d9a62"}
-	contents, err := contentResolver.ResolveContents(diffUuids, tid)
+	contents, err := contentResolver.ResolveContentsNew(diffUuids, tid, requestTimeout)
 	if err != nil {
 		assert.FailNow(t, "Failed retrieving contents.", err.Error())
 	}
 
 	assert.Equal(t, 2, len(contents), "There should be 2 contents retrieved.")
 }
 
+func Test_callContentResolverApp_3_Content(t *testing.T) {
+	mockDSAPI(t, statusWorking, "document-store-api-3-content-output.json")
+
+	diffUuids := []string{"ab43b1a6-1f47-11e7-b7d3-163f5a7f229c", "70c800d8-b3e3-11e6-ba85-95d1533d9a62", "70c800d8-b3e3-11e6-ba85-95d1533d9a63"}
+	contents, err := contentResolver.ResolveContentsNew(diffUuids, tid, requestTimeout)
+	if err != nil {
+		assert.FailNow(t, "Failed retrieving contents.", err.Error())
+	}
+
+	assert.Equal(t, 3, len(contents), "There should be 3 contents retrieved.")
+}
+
 func Test_callContentResolverApp_Empty_Content(t *testing.T) {
 	mockDSAPIBytes(statusWorking, []byte("[]"))
 
 	var diffUuids []string
-	contents, err := contentResolver.ResolveContents(diffUuids, tid)
+	contents, err := contentResolver.ResolveContentsNew(diffUuids, tid, requestTimeout)
 	if err != nil {
 		assert.FailNow(t, "Failed retrieving contents.", err.Error())
 	}
@@ -93,7 +106,7 @@ func Test_callContentResolverApp_NotWorking(t *testing.T) {
 	mockDSAPIBytes(statusNotWorking, []byte("[]"))
 
 	var diffUuids []string
-	_, err := contentResolver.ResolveContents(diffUuids, tid)
+	_, err := contentResolver.ResolveContentsNew(diffUuids, tid, requestTimeout)
 	if err == nil {
 		assert.FailNow(t, "Should have thrown error for failing to reach service.", err.Error())
 	}