Initial implementation

.gitignore

@@ -0,0 +1,5 @@
+*.iml
+*.exe
+.idea
+content-rw-elasticsearch
+vendor/*/

Dockerfile

@@ -0,0 +1,33 @@
+FROM golang:1.8-alpine
+
+ENV PROJECT=content-rw-elasticsearch
+COPY . /${PROJECT}-sources/
+
+RUN apk --no-cache --virtual .build-dependencies add git \
+  && ORG_PATH="github.com/Financial-Times" \
+  && REPO_PATH="${ORG_PATH}/${PROJECT}" \
+  && mkdir -p $GOPATH/src/${ORG_PATH} \
+  # Linking the project sources in the GOPATH folder
+  && ln -s /${PROJECT}-sources $GOPATH/src/${REPO_PATH} \
+  && cd $GOPATH/src/${REPO_PATH} \
+  && BUILDINFO_PACKAGE="${ORG_PATH}/${PROJECT}/vendor/${ORG_PATH}/service-status-go/buildinfo." \
+  && VERSION="version=$(git describe --tag --always 2> /dev/null)" \
+  && DATETIME="dateTime=$(date -u +%Y%m%d%H%M%S)" \
+  && REPOSITORY="repository=$(git config --get remote.origin.url)" \
+  && REVISION="revision=$(git rev-parse HEAD)" \
+  && BUILDER="builder=$(go version)" \
+  && LDFLAGS="-X '"${BUILDINFO_PACKAGE}$VERSION"' -X '"${BUILDINFO_PACKAGE}$DATETIME"' -X '"${BUILDINFO_PACKAGE}$REPOSITORY"' -X '"${BUILDINFO_PACKAGE}$REVISION"' -X '"${BUILDINFO_PACKAGE}$BUILDER"'" \
+  && echo "Build flags: $LDFLAGS" \
+  && echo "Fetching dependencies..." \
+  && go get -u github.com/kardianos/govendor \
+  && $GOPATH/bin/govendor sync \
+  && go get -v \
+  && go build -ldflags="${LDFLAGS}" \
+  && mv ${PROJECT} /${PROJECT} \
+  && mv runtime / \
+  && apk del .build-dependencies \
+  && rm -rf $GOPATH /var/cache/apk/*
+
+WORKDIR /
+# Using the expanded command, so that the shell will expand the $PROJECT env var. See https://docs.docker.com/engine/reference/builder/#cmd
+CMD ["sh", "-c", "/${PROJECT}"]

README.md

@@ -1,2 +1,85 @@
 # content-rw-elasticsearch
-Reader-writer for ES backing replatformed SAPI v1
+
+[![Circle CI](https://circleci.com/gh/Financial-Times/content-rw-elasticsearch/tree/master.png?style=shield)](https://circleci.com/gh/Financial-Times/content-rw-elasticsearch/tree/master)[![Go Report Card](https://goreportcard.com/badge/github.com/Financial-Times/content-rw-elasticsearch)](https://goreportcard.com/report/github.com/Financial-Times/content-rw-elasticsearch) [![Coverage Status](https://coveralls.io/repos/github/Financial-Times/content-rw-elasticsearch/badge.svg)](https://coveralls.io/github/Financial-Times/content-rw-elasticsearch)
+
+
+## Introduction
+Indexes V2 content in Elasticsearch for use by SAPI v1
+
+## Installation
+Download the source code, dependencies and test dependencies:
+
+        go get -u github.com/kardianos/govendor
+        go get -u github.com/Financial-Times/content-rw-elasticsearch
+        cd $GOPATH/src/github.com/Financial-Times/content-rw-elasticsearch
+        govendor sync
+        go get -t
+        go build .
+
+## Running locally
+
+1. Run the tests and install the binary:
+
+        govendor sync
+        govendor test -v -race
+        go install
+
+2. Run the binary (using the `help` flag to see the available optional arguments):
+
+        $GOPATH/bin/content-rw-elasticsearch [--help]
+
+Options:
+
+        --app-system-code="content_rw_elasticsearch"            System Code of the application ($APP_SYSTEM_CODE)
+        --port="8080"                                           Port to listen on ($APP_PORT)
+        --aws-access-key=""                                     AWS ACCES KEY ($AWS_ACCESS_KEY_ID)
+        --aws-secret-access-key=""                              AWS SECRET ACCES KEY ($AWS_SECRET_ACCESS_KEY)
+        --elasticsearch-sapi-endpoint="http://localhost:9200"   AES endpoint ($ELASTICSEARCH_SAPI_ENDPOINT)
+        --index-name="ft"                                       The name of the elaticsearch index ($ELASTICSEARCH_SAPI_INDEX)
+        --kafka-proxy-address="http://localhost:8080"           Addresses used by the queue consumer to connect to the queue ($KAFKA_PROXY_ADDR)
+        --kafka-consumer-group="default-consumer-group"         Group used to read the messages from the queue ($KAFKA_CONSUMER_GROUP)
+        --kafka-topic="CombinedPostPublicationEvents"           The topic to read the meassages from ($KAFKA_TOPIC)
+        --kafka-header="kafka"                                  The header identifying the queue to read the messages from ($KAFKA_HEADER)
+        --kafka-concurrent-processing=false                     Whether the consumer uses concurrent processing for the messages ($KAFKA_CONCURRENT_PROCESSING)
+
+3. Test:
+
+There are no service endpoints to test.
+
+## Build and deployment
+
+* Built by Docker Hub on merge to master: [coco/content-rw-elasticsearch](https://hub.docker.com/r/coco/content-rw-elasticsearch/)
+* CI provided by CircleCI: [content-rw-elasticsearch](https://circleci.com/gh/Financial-Times/content-rw-elasticsearch)
+
+## Healthchecks
+Admin endpoints are:
+
+    `/__gtg`
+
+Returns 503 if any if the checks executed at the /__health endpoint returns false
+
+    `/__health`
+
+There are several checks performed:
+* Elasticsearch cluster connectivity
+* Elasticsearch cluster health
+* Elastic schema validation
+* Kafka queue topic check
+
+
+    `/__health-details`
+
+Shows ES cluster health details
+
+    `/__build-info` 
+
+
+## Other information
+An example of event structure is here [testdata/exampleEnrichedContentModel.json](testdata/exampleEnrichedContentModel.json)
+
+The reference mappings for Elasticsearch are found here [runtime/referenceSchema.json](runtime/referenceSchema.json)
+
+### Logging
+
+* The application uses [logrus](https://github.com/Sirupsen/logrus); the log file is initialised in [main.go](main.go).
+* NOTE: `/__build-info` and `/__gtg` endpoints are not logged as they are called every second from varnish/vulcand and this information is not needed in logs/splunk.

circle.yml

@@ -0,0 +1,31 @@
+machine:
+  environment:
+    GOPATH: "${HOME}/.go_workspace:/usr/local/go_workspace:${HOME}/.go_project"
+    GO_PROJECT_PATH: "${HOME}/.go_project/src/github.com/${CIRCLE_PROJECT_USERNAME}/${CIRCLE_PROJECT_REPONAME}"
+
+checkout:
+  post:
+    - mkdir -p $(dirname ${GO_PROJECT_PATH})
+    - ln -fs $HOME/$CIRCLE_PROJECT_REPONAME $GO_PROJECT_PATH
+
+dependencies:
+  pre:
+    - go get -u github.com/kardianos/govendor
+  override:
+    - cd $GO_PROJECT_PATH && govendor sync
+    - cd $GO_PROJECT_PATH && go build -v
+
+test:
+  pre:
+    - go get -u github.com/jstemmer/go-junit-report
+    - go get -u github.com/mattn/goveralls
+  override:
+    - mkdir -p $CIRCLE_TEST_REPORTS/golang
+    - cd $GO_PROJECT_PATH && govendor test -race -v +local | go-junit-report > $CIRCLE_TEST_REPORTS/golang/junit.xml
+    - cd $GO_PROJECT_PATH && govendor test -v -cover -race -coverprofile=$CIRCLE_ARTIFACTS/coverage.out +local
+    - cd $CIRCLE_ARTIFACTS && sed -i '1d' *.out
+    - |
+      echo "mode: atomic" > $CIRCLE_ARTIFACTS/overall-coverage.result
+    - cd $CIRCLE_ARTIFACTS && cat *.out >> overall-coverage.result
+  post:
+    - goveralls -coverprofile=$CIRCLE_ARTIFACTS/overall-coverage.result -service=circle-ci -repotoken=$COVERALLS_TOKEN

content_indexer.go

@@ -0,0 +1,176 @@
+package main
+
+import (
+	"encoding/json"
+	health "github.com/Financial-Times/go-fthealth/v1_1"
+	"github.com/Financial-Times/http-handlers-go/httphandlers"
+	"github.com/Financial-Times/message-queue-gonsumer/consumer"
+	status "github.com/Financial-Times/service-status-go/httphandlers"
+	log "github.com/Sirupsen/logrus"
+	"github.com/gorilla/mux"
+	"github.com/kr/pretty"
+	"github.com/rcrowley/go-metrics"
+	"net"
+	"net/http"
+	"os"
+	"os/signal"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+)
+
+const syntheticRequestPrefix = "SYNTHETIC-REQ-MON"
+
+type contentIndexer struct {
+	esServiceInstance esServiceI
+}
+
+func waitForSignal() {
+	ch := make(chan os.Signal)
+	signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
+	<-ch
+}
+
+func (indexer *contentIndexer) start(appSystemCode string, indexName string, port string, accessConfig esAccessConfig, queueConfig consumer.QueueConfig) {
+	channel := make(chan esClientI)
+	go func() {
+		defer close(channel)
+		for {
+			ec, err := newAmazonClient(accessConfig)
+			if err == nil {
+				log.Infof("connected to ElasticSearch")
+				channel <- ec
+				return
+			}
+			log.Errorf("could not connect to ElasticSearch: %s", err.Error())
+			time.Sleep(time.Minute)
+		}
+	}()
+
+	//create writer service
+	indexer.esServiceInstance = newEsService(indexName)
+
+	go func() {
+		for ec := range channel {
+			indexer.esServiceInstance.setClient(ec)
+			indexer.startMessageConsumer(queueConfig)
+		}
+	}()
+
+	go func() {
+		indexer.serveAdminEndpoints(appSystemCode, port, queueConfig)
+	}()
+}
+
+func (indexer *contentIndexer) serveAdminEndpoints(appSystemCode string, port string, queueConfig consumer.QueueConfig) {
+	healthService := newHealthService(indexer.esServiceInstance, queueConfig.Topic, queueConfig.Addrs[0])
+	var monitoringRouter http.Handler = mux.NewRouter()
+	monitoringRouter = httphandlers.TransactionAwareRequestLoggingHandler(log.StandardLogger(), monitoringRouter)
+	monitoringRouter = httphandlers.HTTPMetricsHandler(metrics.DefaultRegistry, monitoringRouter)
+
+	serveMux := http.NewServeMux()
+
+	hc := health.HealthCheck{SystemCode: appSystemCode, Name: appSystemCode, Description: "Content Read Writer for Elasticsearch", Checks: healthService.checks}
+	serveMux.HandleFunc("/__health", health.Handler(hc))
+	serveMux.HandleFunc("/__health-details", healthService.HealthDetails)
+	serveMux.HandleFunc(status.GTGPath, status.NewGoodToGoHandler(healthService.gtgCheck))
+	serveMux.HandleFunc(status.BuildInfoPath, status.BuildInfoHandler)
+
+	serveMux.Handle("/", monitoringRouter)
+
+	if err := http.ListenAndServe(":"+port, serveMux); err != nil {
+		log.Fatalf("Unable to start: %v", err)
+	}
+}
+
+func (indexer *contentIndexer) startMessageConsumer(config consumer.QueueConfig) {
+	client := &http.Client{
+		Transport: &http.Transport{
+			Proxy: http.ProxyFromEnvironment,
+			DialContext: (&net.Dialer{
+				Timeout:   30 * time.Second,
+				KeepAlive: 30 * time.Second,
+			}).DialContext,
+			MaxIdleConnsPerHost:   20,
+			TLSHandshakeTimeout:   3 * time.Second,
+			ExpectContinueTimeout: 1 * time.Second,
+		},
+	}
+	messageConsumer := consumer.NewConsumer(config, indexer.handleMessage, client)
+	log.Printf("[Startup] Consumer: %# v", pretty.Formatter(messageConsumer))
+
+	var consumerWaitGroup sync.WaitGroup
+	consumerWaitGroup.Add(1)
+
+	go func() {
+		messageConsumer.Start()
+		consumerWaitGroup.Done()
+	}()
+
+	waitForSignal()
+	messageConsumer.Stop()
+	consumerWaitGroup.Wait()
+}
+
+func (indexer *contentIndexer) handleMessage(msg consumer.Message) {
+
+	tid := msg.Headers["X-Request-Id"]
+
+	if strings.Contains(tid, syntheticRequestPrefix) {
+		log.Infof("[%s] Ignoring synthetic message", tid)
+		return
+	}
+
+	var combinedPostPublicationEvent enrichedContentModel
+	err := json.Unmarshal([]byte(msg.Body), &combinedPostPublicationEvent)
+	if err != nil {
+		log.Errorf("[%s] Cannot unmarshal message body:[%v]", tid, err.Error())
+		return
+	}
+
+	uuid := combinedPostPublicationEvent.Content.UUID
+	log.Printf("[%s] Processing combined post publication event for uuid [%s]", tid, uuid)
+
+	var contentType string
+
+	for _, identifier := range combinedPostPublicationEvent.Content.Identifiers {
+		if strings.HasPrefix(identifier.Authority, "http://api.ft.com/system/FT-LABS-WP") {
+			contentType = "blogPost"
+		} else if strings.HasPrefix(identifier.Authority, "http://api.ft.com/system/FTCOM-METHODE") {
+			contentType = "article"
+		} else if strings.HasPrefix(identifier.Authority, "http://api.ft.com/system/BRIGHTCOVE") {
+			contentType = "video"
+		}
+	}
+
+	if contentType == "" {
+		origin := msg.Headers["Origin-System-Id"]
+		if strings.Contains(origin, "methode-web-pub") {
+			contentType = "article"
+		} else if strings.Contains(origin, "wordpress") {
+			contentType = "blogPost"
+		} else if strings.Contains(origin, "brightcove") {
+			contentType = "video"
+		} else {
+			log.Errorf("[%s] Failed to index content with UUID %s. Could not infer type of content.", tid, uuid)
+			return
+		}
+	}
+
+	if combinedPostPublicationEvent.Content.MarkedDeleted {
+		_, err = indexer.esServiceInstance.deleteData(contentTypeMap[contentType].collection, uuid)
+		if err != nil {
+			log.Errorf(err.Error())
+			return
+		}
+	} else {
+		payload := convertToESContentModel(combinedPostPublicationEvent, contentType)
+
+		_, err = indexer.esServiceInstance.writeData(contentTypeMap[contentType].collection, uuid, payload)
+		if err != nil {
+			log.Errorf(err.Error())
+			return
+		}
+	}
+}