diff --git a/README b/README
deleted file mode 100644
index 09fec22..0000000
--- a/README
+++ /dev/null
@@ -1,3 +0,0 @@
-# apidApigeeAnalytics
-
-This core plugin for [apid](http://github.com/30x/apid) and resposible for collecting analytics data for runtime traffic and puplishing to Apigee.
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..0b41867
--- /dev/null
+++ b/README.md
@@ -0,0 +1,56 @@
+# ApidAnalytics
+
+This is a core plugin for [apid](http://github.com/30x/apid) and is responsible for collecting analytics data for
+runtime traffic from Micro and Enterprise Gateway and puplishing to Apigee.
+
+### Configuration
+
+| name                                  | description                       |
+|---------------------------------------|-----------------------------------|
+| apidanalytics_base_path               | string. default: /analytics       |
+| apidanalytics_data_path               | string. default: /ax              |
+| apidanalytics_collection_interval     | int. seconds. default: 120        |
+| apidanalytics_upload_interval         | int. seconds. default: 5          |
+| apidanalytics_uap_server_base         | string. url. required.            |
+| apidanalytics_use_caching             | boolean. default: true            |
+| apidanalytics_buffer_channel_size     | int. number of slots. default: 100|
+
+### Startup Procedure
+1. Initialize crash recovery, upload and buffering manager to handle buffering analytics messages to files
+   locally and then periodically upload these files to S3/GCS based on signedURL received from
+   uapCollectionEndpoint exposed via edgex proxy
+2. Create a listener for Apigee-Sync event
+    1. Each time a Snapshot is received, create an in-memory cache for data scope and developer information
+    2. Each time a changeList is received
+        1. if data_scope info changed, then insert/delete info for changed scope from tenantCache
+        2. if any other kms table is changed, then refresh entire developerInfo cache as only specific
+           fields are saved in the cache
+3. Initialize POST /analytics/{scope_uuid} API
+4. Upon receiving POST requests
+    1. Validate and enrich each batch of analytics records
+    2. If valid, then publish records to an internal buffer channel
+5. Buffering Logic
+    1. Buffering manager creates listener on the internal buffer channel and thus consumes messages
+       as soon as they are put on the channel
+    2. Based on the current timestamp either an existing directory is used to save these messages
+       or new a new timestamp directory is created
+    3. If a new directory is created, then an event will be published on the closeBucketEvent Channel
+       at the expected directory closing time
+    4. The messages are stored in a file under tmp/<timestamp_directory>
+    5. Based on collection interval, periodically the files in tmp are closed by the routine listening on the
+       closeBucketEvent channel and the directory is moved to staging directory
+6. Upload Manager
+    1. The upload manager periodically checks the staging directory to look for new folders
+    2. When a new folder arrives here, it means all files under that are closed and ready to uploaded
+    3. Tenant info is extracted from the directory name and the files are sequentially uploaded to S3/GCS
+    4. Based on the upload status
+        1. If upload is successful then directory is deleted from staging and previously failed uploads are retried
+        2. if upload fails, then upload is retried 3 times before moving the directory to failed directory
+7. Crash Recovery is a one time activity performed when the plugin is started to
+   cleanly handle open files from a previous Apid stop or crash event
+
+### Exposed API
+```sh
+POST /analytics/{bundle_scope_uuid}
+```
+Complete spec is listed in  `api.yaml`
diff --git a/api.go b/api.go
new file mode 100644
index 0000000..03f5e31
--- /dev/null
+++ b/api.go
@@ -0,0 +1,67 @@
+package apidAnalytics
+
+import (
+	"github.com/30x/apid"
+	"net/http"
+	"strings"
+)
+
+var analyticsBasePath string
+
+type errResponse struct {
+	ErrorCode string `json:"errorCode"`
+	Reason    string `json:"reason"`
+}
+
+type dbError struct {
+	ErrorCode string `json:"errorCode"`
+	Reason    string `json:"reason"`
+}
+
+func initAPI(services apid.Services) {
+	log.Debug("initialized API's exposed by apidAnalytics plugin")
+	analyticsBasePath = config.GetString(configAnalyticsBasePath)
+	services.API().HandleFunc(analyticsBasePath+"/{bundle_scope_uuid}",
+		saveAnalyticsRecord).Methods("POST")
+}
+
+func saveAnalyticsRecord(w http.ResponseWriter, r *http.Request) {
+
+	w.Header().Set("Content-Type", "application/json; charset=UTF-8")
+
+	db := getDB() // When database isnt initialized
+	if db == nil {
+		writeError(w, http.StatusInternalServerError,
+			"INTERNAL_SERVER_ERROR",
+			"Service is not initialized completely")
+		return
+	}
+
+	if !strings.EqualFold(r.Header.Get("Content-Type"), "application/json") {
+		writeError(w, http.StatusBadRequest, "UNSUPPORTED_CONTENT_TYPE",
+			"Only supported content type is application/json")
+		return
+	}
+
+	vars := apid.API().Vars(r)
+	scopeuuid := vars["bundle_scope_uuid"]
+	tenant, dbErr := getTenantForScope(scopeuuid)
+	if dbErr.ErrorCode != "" {
+		switch dbErr.ErrorCode {
+		case "INTERNAL_SEARCH_ERROR":
+			writeError(w, http.StatusInternalServerError,
+				"INTERNAL_SEARCH_ERROR", dbErr.Reason)
+		case "UNKNOWN_SCOPE":
+			writeError(w, http.StatusBadRequest,
+				"UNKNOWN_SCOPE", dbErr.Reason)
+		}
+	} else {
+		err := processPayload(tenant, scopeuuid, r)
+		if err.ErrorCode == "" {
+			w.WriteHeader(http.StatusOK)
+		} else {
+			writeError(w, http.StatusBadRequest,
+				err.ErrorCode, err.Reason)
+		}
+	}
+}
diff --git a/api.yaml b/api.yaml
new file mode 100644
index 0000000..9b9ccfa
--- /dev/null
+++ b/api.yaml
@@ -0,0 +1,147 @@
+swagger: "2.0"
+info:
+  version: "v1"
+  title: APID analytics API
+host: playground.apistudio.io
+basePath: /try/64e409ad-aebb-4bbc-977e-f0e0f22209d4
+schemes:
+  - http
+  - https
+consumes:
+  - application/json
+produces:
+  - application/json
+paths:
+  '/analytics/{bundle_scope_uuid}':
+    x-swagger-router-controller: analytics
+    parameters:
+      - name: bundle_scope_uuid
+        in: path
+        required: true
+        description: bundle UUID that can be mapped to a scope by APID
+        type: string
+      - name: analytics_data
+        in: body
+        description: The analytics data you want to post
+        required: true
+        schema:
+          $ref: "#/definitions/records"
+    post:
+      responses:
+        "200":
+          description: Success
+        "400":
+          description: Bad Request
+          schema:
+            $ref: "#/definitions/errClientError"
+        "500":
+          description: Server error
+          schema:
+            $ref: "#/definitions/errServerError"
+        default:
+          description: Error
+          schema:
+            $ref: "#/definitions/errResponse"
+
+definitions:
+  records:
+    type: object
+    required:
+      - records
+    properties:
+      records:
+        type: array
+        minItems: 1
+        items:
+          $ref: "#/definitions/eachRecord"
+    example: {
+    "records":[{
+        "response_status_code": 400,
+        "client_received_start_timestamp": 1462850097576,
+        "client_id":"0GJKn7EQmNkKYcGL7x3gHaawWLs5gUPr"
+      },{
+        "response_status_code": 200,
+        "client_id":"2ngXgr6Rl2PXWiEmbt8zCkWY3Ptjb8ep",
+        "request_verb" : "GET",
+        "api_product":" test_product",
+        "access_token" : "fewGWG343LDV346345YCDS",
+        "apiproxy" : "OAuthProxy",
+        "apiproxy_revision" : "2",
+        "client_ip": "10.16.9.11",
+        "client_sent_end_timestamp": 1462850097894,
+        "client_received_start_timestamp": 1462850097576,
+      	"client_received_end_timestamp": 1462850097580,
+      	"client_sent_start_timestamp": 1462850097894,
+        "request_path" : "/oauth/oauthv2/auth_code/",
+        "request_uri": "/oauth/oauthv2/auth_code/?response_type=code&redirect_url=http%3A%2F%2Fexample.com&client_id=A1h6yYAVeADnEKji8M37zCSn6olcmQDB",
+        "useragent" : "Chrome",
+        "target" : "target_name",
+        "target_received_end_timestamp": 1462850097800,
+        "target_received_start_timestamp": 1462850097800,
+        "target_response_code" : 200,
+        "target_sent_end_timestamp" : 1462850097802,
+        "target_sent_start_timestamp" :  1462850097802
+      }]
+    }
+
+  eachRecord:
+    description: Each record is basically a map of key value pair. client_received_start_timestamp is a required property but more fields can be added
+    type: object
+    required:
+      - client_received_start_timestamp
+    properties:
+      client_received_start_timestamp:
+        type: integer
+        format: int64
+    example: {
+      "response_status_code":400,
+      "client_received_start_timestamp":1462850097576,
+      "client_id":"0GJKn7EQmNkKYcGL7x3gHaawWLs5gUPr"
+    }
+
+  errClientError:
+    required:
+      - errrorCode
+      - reason
+    properties:
+      errrorCode:
+        type: string
+        enum:
+          - UNKNOWN_SCOPE
+          - BAD_DATA
+          - UNSUPPORTED_CONTENT_TYPE
+          - UNSUPPORTED_CONTENT_ENCODING
+          - MISSING_FIELD
+      reason:
+        type: string
+    example: {
+      "errrorCode":"UNKNOWN_SCOPE",
+      "reason":"No tenant found for this scopeuuid : UUID"
+    }
+
+  errServerError:
+    required:
+      - errrorCode
+      - reason
+    properties:
+      errrorCode:
+        type: string
+        enum:
+          - INTERNAL_SERVER_ERROR
+          - INTERNAL_SEARCH_ERROR
+      reason:
+        type: string
+    example: {
+      "errrorCode":"INTERNAL_SERVER_ERROR",
+      "reason":"Service is not initialized completely"
+    }
+
+  errResponse:
+    required:
+      - errrorCode
+      - reason
+    properties:
+      errrorCode:
+        type: string
+      reason:
+        type: string
\ No newline at end of file
diff --git a/api_helper.go b/api_helper.go
new file mode 100644
index 0000000..24dba22
--- /dev/null
+++ b/api_helper.go
@@ -0,0 +1,181 @@
+package apidAnalytics
+
+import (
+	"compress/gzip"
+	"encoding/json"
+	"io"
+	"net/http"
+	"strings"
+)
+
+/*
+Implements all the helper methods needed to process the POST /analytics payload
+and send it to the internal buffer channel
+*/
+
+type developerInfo struct {
+	ApiProduct     string
+	DeveloperApp   string
+	DeveloperEmail string
+	Developer      string
+}
+
+type axRecords struct {
+	Tenant tenant
+	// Records is an array of multiple analytics records
+	Records []interface{}
+}
+
+type tenant struct {
+	Org      string
+	Env      string
+	TenantId string
+}
+
+func processPayload(tenant tenant, scopeuuid string, r *http.Request) errResponse {
+	var gzipEncoded bool
+	if r.Header.Get("Content-Encoding") != "" {
+		if !strings.EqualFold(r.Header.Get("Content-Encoding"), "gzip") {
+			return errResponse{
+				ErrorCode: "UNSUPPORTED_CONTENT_ENCODING",
+				Reason:    "Only supported content encoding is gzip"}
+		} else {
+			gzipEncoded = true
+		}
+	}
+
+	var reader io.ReadCloser
+	var err error
+	if gzipEncoded {
+		reader, err = gzip.NewReader(r.Body) // reader for gzip encoded data
+		if err != nil {
+			return errResponse{
+				ErrorCode: "BAD_DATA",
+				Reason:    "Gzip Encoded data cannot be read"}
+		}
+	} else {
+		reader = r.Body
+	}
+
+	errMessage := validateEnrichPublish(tenant, scopeuuid, reader)
+	if errMessage.ErrorCode != "" {
+		return errMessage
+	}
+	return errResponse{}
+}
+
+func validateEnrichPublish(tenant tenant, scopeuuid string, reader io.Reader) errResponse {
+	var raw map[string]interface{}
+	decoder := json.NewDecoder(reader) // Decode payload to JSON data
+	decoder.UseNumber()
+
+	if err := decoder.Decode(&raw); err != nil {
+		return errResponse{ErrorCode: "BAD_DATA",
+			Reason: "Not a valid JSON payload"}
+	}
+
+	if records := raw["records"]; records != nil {
+		// Iterate through each record to validate and enrich it
+		for _, eachRecord := range records.([]interface{}) {
+			recordMap := eachRecord.(map[string]interface{})
+			valid, err := validate(recordMap)
+			if valid {
+				enrich(recordMap, scopeuuid, tenant)
+			} else {
+				// Even if there is one bad record, then reject entire batch
+				return err
+			}
+		}
+		axRecords := axRecords{
+			Tenant:  tenant,
+			Records: records.([]interface{})}
+		// publish batch of records to channel (blocking call)
+		internalBuffer <- axRecords
+	} else {
+		return errResponse{
+			ErrorCode: "NO_RECORDS",
+			Reason:    "No analytics records in the payload"}
+	}
+	return errResponse{}
+}
+
+/*
+Does basic validation on each analytics message
+1. client_received_start_timestamp should exist
+2. if client_received_end_timestamp exists then it should be > client_received_start_timestamp
+*/
+func validate(recordMap map[string]interface{}) (bool, errResponse) {
+	elems := []string{"client_received_start_timestamp"}
+	for _, elem := range elems {
+		if recordMap[elem] == nil {
+			return false, errResponse{
+				ErrorCode: "MISSING_FIELD",
+				Reason:    "Missing Required field: " + elem}
+		}
+	}
+
+	crst, exists1 := recordMap["client_received_start_timestamp"]
+	cret, exists2 := recordMap["client_received_end_timestamp"]
+	if exists1 && exists2 {
+		if crst.(json.Number) > cret.(json.Number) {
+			return false, errResponse{
+				ErrorCode: "BAD_DATA",
+				Reason: "client_received_start_timestamp " +
+					"> client_received_end_timestamp"}
+		}
+	}
+	return true, errResponse{}
+}
+
+/*
+Enrich each record by adding org and env fields
+It also finds add developer related information based on the apiKey
+*/
+func enrich(recordMap map[string]interface{}, scopeuuid string, tenant tenant) {
+	org, orgExists := recordMap["organization"]
+	if !orgExists || org.(string) == "" {
+		recordMap["organization"] = tenant.Org
+	}
+
+	env, envExists := recordMap["environment"]
+	if !envExists || env.(string) == "" {
+		recordMap["environment"] = tenant.Env
+	}
+
+	apiKey, exists := recordMap["client_id"]
+	// apiKey doesnt exist then ignore adding developer fields
+	if exists {
+		apiKey := apiKey.(string)
+		devInfo := getDeveloperInfo(tenant.TenantId, apiKey)
+		_, exists := recordMap["api_product"]
+		if !exists {
+			recordMap["api_product"] = devInfo.ApiProduct
+		}
+		_, exists = recordMap["developer_app"]
+		if !exists {
+			recordMap["developer_app"] = devInfo.DeveloperApp
+		}
+		_, exists = recordMap["developer_email"]
+		if !exists {
+			recordMap["developer_email"] = devInfo.DeveloperEmail
+		}
+		_, exists = recordMap["developer"]
+		if !exists {
+			recordMap["developer"] = devInfo.Developer
+		}
+	}
+}
+
+func writeError(w http.ResponseWriter, status int, code string, reason string) {
+	w.WriteHeader(status)
+	e := errResponse{
+		ErrorCode: code,
+		Reason:    reason,
+	}
+	bytes, err := json.Marshal(e)
+	if err != nil {
+		log.Errorf("unable to marshal errorResponse: %v", err)
+	} else {
+		w.Write(bytes)
+	}
+}
diff --git a/api_test.go b/api_test.go
new file mode 100644
index 0000000..d172a29
--- /dev/null
+++ b/api_test.go
@@ -0,0 +1,49 @@
+package apidAnalytics
+
+import (
+	"net/http"
+	"net/url"
+	"strings"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+// BeforeSuite setup and AfterSuite cleanup is in apidAnalytics_suite_test.go
+var _ = Describe("testing saveAnalyticsRecord() directly", func() {
+	Context("valid scopeuuid", func() {
+		It("should successfully return", func() {
+			uri, err := url.Parse(testServer.URL)
+			uri.Path = analyticsBasePath
+
+			v := url.Values{}
+			v.Add("bundle_scope_uuid", "testid")
+
+			client := &http.Client{}
+			req, err := http.NewRequest("POST", uri.String(),
+				strings.NewReader(v.Encode()))
+			res, err := client.Do(req)
+			defer res.Body.Close()
+			Expect(err).ShouldNot(HaveOccurred())
+			Expect(res.StatusCode, http.StatusOK)
+		})
+	})
+
+	Context("invalid scopeuuid", func() {
+		It("should return bad request", func() {
+			uri, err := url.Parse(testServer.URL)
+			uri.Path = analyticsBasePath
+
+			v := url.Values{}
+			v.Add("bundle_scope_uuid", "wrongId")
+
+			client := &http.Client{}
+			req, err := http.NewRequest("POST", uri.String(),
+				strings.NewReader(v.Encode()))
+			res, err := client.Do(req)
+			defer res.Body.Close()
+			Expect(err).ShouldNot(HaveOccurred())
+			Expect(res.StatusCode, http.StatusBadRequest)
+		})
+	})
+})
diff --git a/apidAnalytics_suite_test.go b/apidAnalytics_suite_test.go
new file mode 100644
index 0000000..1e5c128
--- /dev/null
+++ b/apidAnalytics_suite_test.go
@@ -0,0 +1,232 @@
+package apidAnalytics
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/30x/apid"
+	"github.com/30x/apid/factory"
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+)
+
+var (
+	testTempDir string
+	testServer  *httptest.Server
+)
+
+func TestApidAnalytics(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "ApidAnalytics Suite")
+}
+
+var _ = BeforeSuite(func() {
+	apid.Initialize(factory.DefaultServicesFactory())
+
+	config := apid.Config()
+
+	var err error
+	testTempDir, err = ioutil.TempDir("", "api_test")
+	Expect(err).NotTo(HaveOccurred())
+
+	config.Set("data_path", testTempDir)
+	config.Set(uapServerBase, "http://localhost:9000") // dummy value
+	config.Set("apigeesync_apid_instance_id",
+		"abcdefgh-ijkl-mnop-qrst-uvwxyz123456") // dummy value
+	config.Set(useCaching, true)
+
+	db, err := apid.Data().DB()
+	Expect(err).NotTo(HaveOccurred())
+	setDB(db)
+	createApidClusterTables(db)
+	createTables(db)
+	insertTestData(db)
+	apid.InitializePlugins()
+
+	// Create cache else its created in listener.go when a snapshot is received
+	createTenantCache()
+	createDeveloperInfoCache()
+
+	testServer = httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, req *http.Request) {
+			if req.URL.Path == analyticsBasePathDefault {
+				saveAnalyticsRecord(w, req)
+			}
+		}))
+})
+
+func createTables(db apid.DB) {
+	_, err := db.Exec(`
+		CREATE TABLE IF NOT EXISTS api_product (
+		    id text,
+		    tenant_id text,
+		    name text,
+		    display_name text,
+		    description text,
+		    api_resources text[],
+		    approval_type text,
+		    _change_selector text,
+		    proxies text[],
+		    environments text[],
+		    quota text,
+		    quota_time_unit text,
+		    quota_interval int,
+		    created_at int64,
+		    created_by text,
+		    updated_at int64,
+		    updated_by text,
+		    PRIMARY KEY (tenant_id, id));
+		CREATE TABLE IF NOT EXISTS developer (
+		    id text,
+		    tenant_id text,
+		    username text,
+		    first_name text,
+		    last_name text,
+		    password text,
+		    email text,
+		    status text,
+		    encrypted_password text,
+		    salt text,
+		    _change_selector text,
+		    created_at int64,
+		    created_by text,
+		    updated_at int64,
+		    updated_by text,
+		    PRIMARY KEY (tenant_id, id)
+		);
+		CREATE TABLE IF NOT EXISTS app (
+		    id text,
+		    tenant_id text,
+		    name text,
+		    display_name text,
+		    access_type text,
+		    callback_url text,
+		    status text,
+		    app_family text,
+		    company_id text,
+		    developer_id text,
+		    type int,
+		    created_at int64,
+		    created_by text,
+		    updated_at int64,
+		    updated_by text,
+		    _change_selector text,
+		    PRIMARY KEY (tenant_id, id)
+		);
+		CREATE TABLE IF NOT EXISTS app_credential_apiproduct_mapper (
+		    tenant_id text,
+		    appcred_id text,
+		    app_id text,
+		    apiprdt_id text,
+		    _change_selector text,
+		    status text,
+		    PRIMARY KEY (appcred_id, app_id, apiprdt_id,tenant_id)
+		);
+	`)
+	if err != nil {
+		panic("Unable to initialize DB " + err.Error())
+	}
+}
+
+func createApidClusterTables(db apid.DB) {
+	_, err := db.Exec(`
+		CREATE TABLE apid_cluster (
+		    id text,
+		    instance_id text,
+		    name text,
+		    description text,
+		    umbrella_org_app_name text,
+		    created int64,
+		    created_by text,
+		    updated int64,
+		    updated_by text,
+		    _change_selector text,
+		    snapshotInfo text,
+		    lastSequence text,
+		    PRIMARY KEY (id)
+		);
+		CREATE TABLE data_scope (
+		    id text,
+		    apid_cluster_id text,
+		    scope text,
+		    org text,
+		    env text,
+		    created int64,
+		    created_by text,
+		    updated int64,
+		    updated_by text,
+		    _change_selector text,
+		    PRIMARY KEY (id)
+		);
+	`)
+	if err != nil {
+		panic("Unable to initialize DB " + err.Error())
+	}
+}
+
+func insertTestData(db apid.DB) {
+
+	txn, err := db.Begin()
+	Expect(err).ShouldNot(HaveOccurred())
+	txn.Exec("INSERT INTO APP_CREDENTIAL_APIPRODUCT_MAPPER (tenant_id,"+
+		" appcred_id, app_id, apiprdt_id, status, _change_selector) "+
+		"VALUES"+
+		"($1,$2,$3,$4,$5,$6)",
+		"tenantid",
+		"testapikey",
+		"testappid",
+		"testproductid",
+		"APPROVED",
+		"12345",
+	)
+
+	txn.Exec("INSERT INTO APP (id, tenant_id, name, developer_id) "+
+		"VALUES"+
+		"($1,$2,$3,$4)",
+		"testappid",
+		"tenantid",
+		"testapp",
+		"testdeveloperid",
+	)
+
+	txn.Exec("INSERT INTO API_PRODUCT (id, tenant_id, name) "+
+		"VALUES"+
+		"($1,$2,$3)",
+		"testproductid",
+		"tenantid",
+		"testproduct",
+	)
+
+	txn.Exec("INSERT INTO DEVELOPER (id, tenant_id, username, email) "+
+		"VALUES"+
+		"($1,$2,$3,$4)",
+		"testdeveloperid",
+		"tenantid",
+		"testdeveloper",
+		"testdeveloper@test.com",
+	)
+
+	txn.Exec("INSERT INTO DATA_SCOPE (id, _change_selector, "+
+		"apid_cluster_id, scope, org, env) "+
+		"VALUES"+
+		"($1,$2,$3,$4,$5,$6)",
+		"testid",
+		"some_change_selector",
+		"some_cluster_id",
+		"tenantid",
+		"testorg",
+		"testenv",
+	)
+	txn.Commit()
+}
+
+var _ = AfterSuite(func() {
+	apid.Events().Close()
+	if testServer != nil {
+		testServer.Close()
+	}
+	os.RemoveAll(testTempDir)
+})
diff --git a/buffering_manager.go b/buffering_manager.go
new file mode 100644
index 0000000..7e28d4b
--- /dev/null
+++ b/buffering_manager.go
@@ -0,0 +1,179 @@
+package apidAnalytics
+
+import (
+	"bufio"
+	"compress/gzip"
+	"crypto/rand"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+)
+
+const fileExtension = ".txt.gz"
+
+// Channel where analytics records are buffered before being dumped to a
+// file as write to file should not performed in the Http Thread
+var internalBuffer chan axRecords
+
+// Channel where close bucket event is published i.e. when a bucket
+// is ready to be closed based on collection interval
+var closeBucketEvent chan bucket
+
+// Map from timestampt to bucket
+var bucketMap map[int64]bucket
+
+type bucket struct {
+	DirName string
+	// We need file handle and writer to close the file
+	FileWriter fileWriter
+}
+
+// This struct will store open file handle and writer to close the file
+type fileWriter struct {
+	file *os.File
+	gw   *gzip.Writer
+	bw   *bufio.Writer
+}
+
+func initBufferingManager() {
+	internalBuffer = make(chan axRecords,
+		config.GetInt(analyticsBufferChannelSize))
+	closeBucketEvent = make(chan bucket)
+	bucketMap = make(map[int64]bucket)
+
+	// Keep polling the internal buffer for new messages
+	go func() {
+		for {
+			records := <-internalBuffer
+			err := save(records)
+			if err != nil {
+				log.Errorf("Could not save %d messages to file"+
+					" due to: %v", len(records.Records), err)
+			}
+		}
+	}()
+
+	// Keep polling the closeEvent channel to see if bucket is ready to be closed
+	go func() {
+		for {
+			bucket := <-closeBucketEvent
+			log.Debugf("Close Event received for bucket: %s",
+				bucket.DirName)
+
+			// close open file
+			closeGzipFile(bucket.FileWriter)
+
+			dirToBeClosed := filepath.Join(localAnalyticsTempDir, bucket.DirName)
+			stagingPath := filepath.Join(localAnalyticsStagingDir, bucket.DirName)
+			// close files in tmp folder and move directory to
+			// staging to indicate its ready for upload
+			err := os.Rename(dirToBeClosed, stagingPath)
+			if err != nil {
+				log.Errorf("Cannot move directory '%s' from"+
+					" tmp to staging folder", bucket.DirName)
+			}
+		}
+	}()
+}
+
+// Save records to correct file based on what timestamp data is being collected for
+func save(records axRecords) error {
+	bucket, err := getBucketForTimestamp(time.Now(), records.Tenant)
+	if err != nil {
+		return err
+	}
+	writeGzipFile(bucket.FileWriter, records.Records)
+	return nil
+}
+
+func getBucketForTimestamp(now time.Time, tenant tenant) (bucket, error) {
+	// first based on current timestamp and collection interval,
+	// determine the timestamp of the bucket
+	ts := now.Unix() / int64(config.GetInt(analyticsCollectionInterval)) * int64(config.GetInt(analyticsCollectionInterval))
+	_, exists := bucketMap[ts]
+	if exists {
+		return bucketMap[ts], nil
+	} else {
+		timestamp := time.Unix(ts, 0).Format(timestampLayout)
+
+		// endtimestamp of bucket = starttimestamp + collectionInterval
+		endTime := time.Unix(ts+int64(config.GetInt(analyticsCollectionInterval)), 0)
+		endtimestamp := endTime.Format(timestampLayout)
+
+		dirName := tenant.Org + "~" + tenant.Env + "~" + timestamp
+		newPath := filepath.Join(localAnalyticsTempDir, dirName)
+		// create dir
+		err := os.Mkdir(newPath, os.ModePerm)
+		if err != nil {
+			return bucket{}, fmt.Errorf("Cannot create directory "+
+				"'%s' to buffer messages '%v'", dirName, err)
+		}
+
+		// create file for writing
+		// Format: <4DigitRandomHex>_<TSStart>.<TSEnd>_<APIDINSTANCEUUID>_writer_0.txt.gz
+		fileName := getRandomHex() + "_" + timestamp + "." +
+			endtimestamp + "_" +
+			config.GetString("apigeesync_apid_instance_id") +
+			"_writer_0" + fileExtension
+		completeFilePath := filepath.Join(newPath, fileName)
+		fw, err := createGzipFile(completeFilePath)
+		if err != nil {
+			return bucket{}, err
+		}
+
+		newBucket := bucket{DirName: dirName, FileWriter: fw}
+		bucketMap[ts] = newBucket
+
+		//Send event to close directory after endTime + 5
+		// seconds to make sure all buffers are flushed to file
+		timer := time.After(endTime.Sub(time.Now()) + time.Second*5)
+		go func() {
+			<-timer
+			closeBucketEvent <- newBucket
+		}()
+		return newBucket, nil
+	}
+}
+
+// 4 digit Hex is prefixed to each filename to improve
+// how s3 partitions the files being uploaded
+func getRandomHex() string {
+	buff := make([]byte, 2)
+	rand.Read(buff)
+	return fmt.Sprintf("%x", buff)
+}
+
+func createGzipFile(s string) (fileWriter, error) {
+	file, err := os.OpenFile(s, os.O_WRONLY|os.O_CREATE, os.ModePerm)
+	if err != nil {
+		return fileWriter{},
+			fmt.Errorf("Cannot create file '%s' "+
+				"to buffer messages '%v'", s, err)
+	}
+	gw := gzip.NewWriter(file)
+	bw := bufio.NewWriter(gw)
+	return fileWriter{file, gw, bw}, nil
+}
+
+func writeGzipFile(fw fileWriter, records []interface{}) {
+	// write each record as a new line to the bufferedWriter
+	for _, eachRecord := range records {
+		s, _ := json.Marshal(eachRecord)
+		_, err := (fw.bw).WriteString(string(s))
+		if err != nil {
+			log.Errorf("Write to file failed '%v'", err)
+		}
+		(fw.bw).WriteString("\n")
+	}
+	// Flush entire batch of records to file vs each message
+	fw.bw.Flush()
+	fw.gw.Flush()
+}
+
+func closeGzipFile(fw fileWriter) {
+	fw.bw.Flush()
+	fw.gw.Close()
+	fw.file.Close()
+}
diff --git a/common_helper.go b/common_helper.go
new file mode 100644
index 0000000..9acc033
--- /dev/null
+++ b/common_helper.go
@@ -0,0 +1,215 @@
+package apidAnalytics
+
+import (
+	"database/sql"
+	"fmt"
+	"sync"
+)
+
+// Cache for scope uuid to org, env and tenantId information
+var tenantCache map[string]tenant
+
+// RW lock for tenant map cache since the cache can be
+// read while its being written to and vice versa
+var tenantCachelock = sync.RWMutex{}
+
+// Cache for apiKey~tenantId to developer related information
+var developerInfoCache map[string]developerInfo
+
+// RW lock for developerInfo map cache since the cache can be
+// read while its being written to and vice versa
+var developerInfoCacheLock = sync.RWMutex{}
+
+// Load data scope information into an in-memory cache so that
+// for each record a DB lookup is not required
+func createTenantCache() error {
+	// Lock before writing to the map as it has multiple readers
+	tenantCachelock.Lock()
+	defer tenantCachelock.Unlock()
+	tenantCache = make(map[string]tenant)
+	var org, env, tenantId, id string
+
+	db := getDB()
+	rows, error := db.Query("SELECT env, org, scope, id FROM DATA_SCOPE")
+
+	if error != nil {
+		return fmt.Errorf("Count not get datascope from "+
+			"DB due to: %v", error)
+	} else {
+		defer rows.Close()
+		for rows.Next() {
+			rows.Scan(&env, &org, &tenantId, &id)
+			tenantCache[id] = tenant{Org: org,
+				Env:      env,
+				TenantId: tenantId}
+		}
+	}
+
+	log.Debugf("Count of data scopes in the cache: %d", len(tenantCache))
+	return nil
+}
+
+// Load data scope information into an in-memory cache so that
+// for each record a DB lookup is not required
+func createDeveloperInfoCache() error {
+	// Lock before writing to the map as it has multiple readers
+	developerInfoCacheLock.Lock()
+	defer developerInfoCacheLock.Unlock()
+	developerInfoCache = make(map[string]developerInfo)
+	var apiProduct, developerApp, developerEmail, developer sql.NullString
+	var tenantId, apiKey string
+
+	db := getDB()
+	sSql := "SELECT mp.tenant_id, mp.appcred_id, ap.name," +
+		" a.name, d.username, d.email " +
+		"FROM APP_CREDENTIAL_APIPRODUCT_MAPPER as mp " +
+		"INNER JOIN API_PRODUCT as ap ON ap.id = mp.apiprdt_id " +
+		"INNER JOIN APP AS a ON a.id = mp.app_id " +
+		"INNER JOIN DEVELOPER as d ON d.id = a.developer_id;"
+	rows, error := db.Query(sSql)
+
+	if error != nil {
+		return fmt.Errorf("Count not get developerInfo "+
+			"from DB due to: %v", error)
+	} else {
+		defer rows.Close()
+		for rows.Next() {
+			rows.Scan(&tenantId, &apiKey, &apiProduct,
+				&developerApp, &developer, &developerEmail)
+
+			keyForMap := getKeyForDeveloperInfoCache(tenantId, apiKey)
+			apiPrd := getValuesIgnoringNull(apiProduct)
+			devApp := getValuesIgnoringNull(developerApp)
+			dev := getValuesIgnoringNull(developer)
+			devEmail := getValuesIgnoringNull(developerEmail)
+
+			developerInfoCache[keyForMap] = developerInfo{
+				ApiProduct:     apiPrd,
+				DeveloperApp:   devApp,
+				DeveloperEmail: devEmail,
+				Developer:      dev}
+		}
+	}
+
+	log.Debugf("Count of apiKey~tenantId combinations "+
+		"in the cache: %d", len(developerInfoCache))
+	return nil
+}
+
+// Returns Tenant Info given a scope uuid from the cache or by querying
+// the DB directly based on useCachig config
+func getTenantForScope(scopeuuid string) (tenant, dbError) {
+	if config.GetBool(useCaching) {
+		// acquire a read lock as this cache has 1 writer as well
+		tenantCachelock.RLock()
+		defer tenantCachelock.RUnlock()
+		ten, exists := tenantCache[scopeuuid]
+		if !exists {
+			reason := "No tenant found for this scopeuuid: " + scopeuuid
+			errorCode := "UNKNOWN_SCOPE"
+			// Incase of unknown scope, try to refresh the
+			// cache ansynchronously incase an update was missed or delayed
+			go createTenantCache()
+			return tenant{}, dbError{
+				ErrorCode: errorCode,
+				Reason:    reason}
+		} else {
+			return ten, dbError{}
+		}
+	} else {
+		var org, env, tenantId string
+
+		db := getDB()
+		error := db.QueryRow("SELECT env, org, scope FROM DATA_SCOPE"+
+			" where id = ?", scopeuuid).Scan(&env, &org, &tenantId)
+
+		switch {
+		case error == sql.ErrNoRows:
+			reason := "No tenant found for this scopeuuid: " + scopeuuid
+			errorCode := "UNKNOWN_SCOPE"
+			return tenant{}, dbError{
+				ErrorCode: errorCode,
+				Reason:    reason}
+		case error != nil:
+			reason := error.Error()
+			errorCode := "INTERNAL_SEARCH_ERROR"
+			return tenant{}, dbError{
+				ErrorCode: errorCode,
+				Reason:    reason}
+		}
+		return tenant{
+			Org:      org,
+			Env:      env,
+			TenantId: tenantId}, dbError{}
+	}
+}
+
+// Returns Dveloper related info given an apiKey and tenantId
+// from the cache or by querying the DB directly based on useCachig config
+func getDeveloperInfo(tenantId string, apiKey string) developerInfo {
+	if config.GetBool(useCaching) {
+		keyForMap := getKeyForDeveloperInfoCache(tenantId, apiKey)
+		// acquire a read lock as this cache has 1 writer as well
+		tenantCachelock.RLock()
+		defer tenantCachelock.RUnlock()
+		devInfo, exists := developerInfoCache[keyForMap]
+		if !exists {
+			log.Warnf("No data found for for tenantId = %s"+
+				" and apiKey = %s", tenantId, apiKey)
+			// Incase of unknown apiKey~tenantId,
+			// try to refresh the cache ansynchronously incase an update was missed or delayed
+			go createTenantCache()
+			return developerInfo{}
+		} else {
+			return devInfo
+		}
+	} else {
+		var apiProduct, developerApp, developerEmail sql.NullString
+		var developer sql.NullString
+
+		db := getDB()
+		sSql := "SELECT ap.name, a.name, d.username, d.email " +
+			"FROM APP_CREDENTIAL_APIPRODUCT_MAPPER as mp " +
+			"INNER JOIN API_PRODUCT as ap ON ap.id = mp.apiprdt_id " +
+			"INNER JOIN APP AS a ON a.id = mp.app_id " +
+			"INNER JOIN DEVELOPER as d ON d.id = a.developer_id " +
+			"where mp.tenant_id = ? and mp.appcred_id = ?;"
+		error := db.QueryRow(sSql, tenantId, apiKey).
+			Scan(&apiProduct, &developerApp,
+				&developer, &developerEmail)
+
+		switch {
+		case error == sql.ErrNoRows:
+			log.Debugf("No data found for for tenantId = %s "+
+				"and apiKey = %s", tenantId, apiKey)
+			return developerInfo{}
+		case error != nil:
+			log.Debugf("No data found for for tenantId = %s and "+
+				"apiKey = %s due to: %v", tenantId, apiKey, error)
+			return developerInfo{}
+		}
+
+		apiPrd := getValuesIgnoringNull(apiProduct)
+		devApp := getValuesIgnoringNull(developerApp)
+		dev := getValuesIgnoringNull(developer)
+		devEmail := getValuesIgnoringNull(developerEmail)
+		return developerInfo{ApiProduct: apiPrd,
+			DeveloperApp:   devApp,
+			DeveloperEmail: devEmail,
+			Developer:      dev}
+	}
+}
+
+// Helper method to handle scanning null values in DB to empty string
+func getValuesIgnoringNull(sqlValue sql.NullString) string {
+	if sqlValue.Valid {
+		return sqlValue.String
+	} else {
+		return ""
+	}
+}
+
+// Build Key as a combination of tenantId and apiKey for the developerInfo Cache
+func getKeyForDeveloperInfoCache(tenantId string, apiKey string) string {
+	return tenantId + "~" + apiKey
+}
diff --git a/common_helper_test.go b/common_helper_test.go
new file mode 100644
index 0000000..c691d52
--- /dev/null
+++ b/common_helper_test.go
@@ -0,0 +1,48 @@
+package apidAnalytics
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("test getTenantForScope()", func() {
+	Context("get tenant for valid scopeuuid", func() {
+		It("should return testorg and testenv", func() {
+			tenant, dbError := getTenantForScope("testid")
+			Expect(dbError.Reason).To(Equal(""))
+			Expect(tenant.Org).To(Equal("testorg"))
+			Expect(tenant.Env).To(Equal("testenv"))
+			Expect(tenant.TenantId).To(Equal("tenantid"))
+		})
+	})
+
+	Context("get tenant for invalid scopeuuid", func() {
+		It("should return empty tenant and a db error", func() {
+			tenant, dbError := getTenantForScope("wrongid")
+			Expect(tenant.Org).To(Equal(""))
+			Expect(dbError.ErrorCode).To(Equal("UNKNOWN_SCOPE"))
+		})
+	})
+})
+
+var _ = Describe("test getDeveloperInfo()", func() {
+	Context("get developerInfo for valid tenantId and apikey", func() {
+		It("should return all right data", func() {
+			developerInfo := getDeveloperInfo("tenantid", "testapikey")
+			Expect(developerInfo.ApiProduct).To(Equal("testproduct"))
+			Expect(developerInfo.Developer).To(Equal("testdeveloper"))
+			Expect(developerInfo.DeveloperEmail).To(Equal("testdeveloper@test.com"))
+			Expect(developerInfo.DeveloperApp).To(Equal("testapp"))
+		})
+	})
+
+	Context("get developerInfo for invalid tenantId and apikey", func() {
+		It("should return all right data", func() {
+			developerInfo := getDeveloperInfo("wrongid", "wrongapikey")
+			Expect(developerInfo.ApiProduct).To(Equal(""))
+			Expect(developerInfo.Developer).To(Equal(""))
+			Expect(developerInfo.DeveloperEmail).To(Equal(""))
+			Expect(developerInfo.DeveloperApp).To(Equal(""))
+		})
+	})
+})
diff --git a/crash_recovery.go b/crash_recovery.go
new file mode 100644
index 0000000..ccd3b36
--- /dev/null
+++ b/crash_recovery.go
@@ -0,0 +1,189 @@
+package apidAnalytics
+
+import (
+	"bufio"
+	"compress/gzip"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+)
+
+const (
+	crashRecoveryDelay = 30 // seconds
+	// same as "yyyyMMddHHmmss.SSS" format (Appended to Recovered folder and file)
+	recoveryTSLayout = "20060102150405.000"
+	// Constant to identify recovered files
+	recoveredTS = "~recoveredTS~"
+)
+
+func initCrashRecovery() {
+	if crashRecoveryNeeded() {
+		timer := time.After(time.Second * crashRecoveryDelay)
+		// Actual recovery of files is attempted asynchronously
+		// after a delay to not block the apid plugin from starting up
+		go func() {
+			<-timer
+			performRecovery()
+		}()
+	}
+}
+
+// Crash recovery is needed if there are any folders in tmp or recovered directory
+func crashRecoveryNeeded() bool {
+	recoveredDirRecoveryNeeded := recoverFolderInRecoveredDir()
+	tmpDirRecoveryNeeded := recoverFoldersInTmpDir()
+	needed := tmpDirRecoveryNeeded || recoveredDirRecoveryNeeded
+	if needed {
+		log.Infof("Crash Recovery is needed and will be "+
+			"attempted in %d seconds", crashRecoveryDelay)
+	}
+	return needed
+}
+
+// If Apid is shutdown or crashes while a file is still open in tmp folder,
+// then the file has partial data.
+// This partial data can be recoverd.
+func recoverFoldersInTmpDir() bool {
+	tmpRecoveryNeeded := false
+	dirs, _ := ioutil.ReadDir(localAnalyticsTempDir)
+	recoveryTS := getRecoveryTS()
+	for _, dir := range dirs {
+		tmpRecoveryNeeded = true
+		log.Debugf("Moving directory '%s' from tmp to"+
+			" recovered folder", dir.Name())
+		tmpCompletePath := filepath.Join(localAnalyticsTempDir, dir.Name())
+		// Eg. org~env~20160101222400~recoveredTS~20160101222612.123
+		newDirName := dir.Name() + recoveredTS + recoveryTS
+		recoveredCompletePath := filepath.Join(localAnalyticsRecoveredDir, newDirName)
+		err := os.Rename(tmpCompletePath, recoveredCompletePath)
+		if err != nil {
+			log.Errorf("Cannot move directory '%s' "+
+				"from tmp to recovered folder", dir.Name())
+		}
+	}
+	return tmpRecoveryNeeded
+}
+
+// Get Timestamp for when the recovery is being attempted on the folder.
+func getRecoveryTS() string {
+	current := time.Now()
+	return current.Format(recoveryTSLayout)
+}
+
+// If APID is restarted twice immediately such that files have been moved to
+// recovered folder but actual recovery has'nt started or is partially done
+// Then the files will just stay in the recovered dir
+// and need to be recovered again.
+func recoverFolderInRecoveredDir() bool {
+	dirs, _ := ioutil.ReadDir(localAnalyticsRecoveredDir)
+	if len(dirs) > 0 {
+		return true
+	}
+	return false
+}
+
+func performRecovery() {
+	log.Info("Crash recovery is starting...")
+	recoveryDirs, _ := ioutil.ReadDir(localAnalyticsRecoveredDir)
+	for _, dir := range recoveryDirs {
+		recoverDirectory(dir.Name())
+	}
+	log.Info("Crash recovery complete...")
+}
+
+func recoverDirectory(dirName string) {
+	log.Infof("performing crash recovery for directory: %s", dirName)
+	var bucketRecoveryTS string
+
+	// Parse bucket name to extract recoveryTS and pass it each file to be recovered
+	// Eg. org~env~20160101222400~recoveredTS~20160101222612.123
+	// 			-> bucketRecoveryTS = _20160101222612.123
+	index := strings.Index(dirName, recoveredTS)
+	if index != -1 {
+		bucketRecoveryTS = "_" + dirName[index+len(recoveredTS):]
+	}
+
+	dirBeingRecovered := filepath.Join(localAnalyticsRecoveredDir, dirName)
+	files, _ := ioutil.ReadDir(dirBeingRecovered)
+	for _, file := range files {
+		// recovering each file sequentially for now
+		recoverFile(bucketRecoveryTS, dirName, file.Name())
+	}
+
+	stagingPath := filepath.Join(localAnalyticsStagingDir, dirName)
+	err := os.Rename(dirBeingRecovered, stagingPath)
+	if err != nil {
+		log.Errorf("Cannot move directory '%s' from"+
+			" recovered to staging folder", dirName)
+	}
+}
+
+func recoverFile(bucketRecoveryTS, dirName, fileName string) {
+	log.Debugf("performing crash recovery for file: %s ", fileName)
+	// add recovery timestamp to the file name
+	completeOrigFilePath := filepath.Join(localAnalyticsRecoveredDir, dirName, fileName)
+
+	recoveredExtension := "_recovered" + bucketRecoveryTS + fileExtension
+	recoveredFileName := strings.TrimSuffix(fileName, fileExtension) + recoveredExtension
+	// eg. 5be1_20170130155400.20170130155600_218e3d99-efaf-4a7b-b3f2-5e4b00c023b7_writer_0_recovered_20170130155452.616.txt
+	recoveredFilePath := filepath.Join(localAnalyticsRecoveredDir, dirName, recoveredFileName)
+
+	// Copy complete records to new file and delete original partial file
+	copyPartialFile(completeOrigFilePath, recoveredFilePath)
+	deletePartialFile(completeOrigFilePath)
+}
+
+// The file is read line by line and all complete records are extracted
+// and copied to a new file which is closed as a correct gzip file.
+func copyPartialFile(completeOrigFilePath, recoveredFilePath string) {
+	partialFile, err := os.Open(completeOrigFilePath)
+	if err != nil {
+		log.Errorf("Cannot open file: %s", completeOrigFilePath)
+		return
+	}
+	defer partialFile.Close()
+
+	bufReader := bufio.NewReader(partialFile)
+	gzReader, err := gzip.NewReader(bufReader)
+	if err != nil {
+		log.Errorf("Cannot create reader on gzip file: %s"+
+			" due to %v", completeOrigFilePath, err)
+		return
+	}
+	defer gzReader.Close()
+
+	scanner := bufio.NewScanner(gzReader)
+
+	// Create new file to copy complete records from partial file and upload only a complete file
+	recoveredFile, err := os.OpenFile(recoveredFilePath,
+		os.O_WRONLY|os.O_CREATE, os.ModePerm)
+	if err != nil {
+		log.Errorf("Cannot create recovered file: %s", recoveredFilePath)
+		return
+	}
+	defer recoveredFile.Close()
+
+	gzWriter := gzip.NewWriter(recoveredFile)
+	defer gzWriter.Close()
+
+	bufWriter := bufio.NewWriter(gzWriter)
+	defer bufWriter.Flush()
+
+	for scanner.Scan() {
+		bufWriter.Write(scanner.Bytes())
+		bufWriter.WriteString("\n")
+	}
+	if err := scanner.Err(); err != nil {
+		log.Warnf("Error while scanning partial file: %v", err)
+		return
+	}
+}
+
+func deletePartialFile(completeOrigFilePath string) {
+	err := os.Remove(completeOrigFilePath)
+	if err != nil {
+		log.Errorf("Cannot delete partial file: %s", completeOrigFilePath)
+	}
+}
diff --git a/glide.yaml b/glide.yaml
new file mode 100644
index 0000000..054e3eb
--- /dev/null
+++ b/glide.yaml
@@ -0,0 +1,8 @@
+package: github.com/30x/apidAnalytics
+
+import:
+- package: github.com/30x/apid
+  version: master
+testImport:
+- package: github.com/onsi/ginkgo/ginkgo
+- package: github.com/onsi/gomega
\ No newline at end of file
diff --git a/init.go b/init.go
new file mode 100644
index 0000000..a3501a4
--- /dev/null
+++ b/init.go
@@ -0,0 +1,178 @@
+package apidAnalytics
+
+import (
+	"fmt"
+	"github.com/30x/apid"
+	"os"
+	"path/filepath"
+	"sync"
+)
+
+const (
+	// Base path of analytics API that will be exposed
+	configAnalyticsBasePath  = "apidanalytics_base_path"
+	analyticsBasePathDefault = "/analytics"
+
+	// Root directory for analytics local data buffering
+	configAnalyticsDataPath  = "apidanalytics_data_path"
+	analyticsDataPathDefault = "/ax"
+
+	// Data collection and buffering interval in seconds
+	analyticsCollectionInterval        = "apidanalytics_collection_interval"
+	analyticsCollectionIntervalDefault = "120"
+
+	// Interval in seconds based on which staging directory
+	// will be checked for folders ready to be uploaded
+	analyticsUploadInterval        = "apidanalytics_upload_interval"
+	analyticsUploadIntervalDefault = "5"
+
+	// Number of slots for internal channel buffering of
+	// analytics records before they are dumped to a file
+	analyticsBufferChannelSize        = "apidanalytics_buffer_channel_size"
+	analyticsBufferChannelSizeDefault = 100
+
+	// EdgeX endpoint base path to access Uap Collection Endpoint
+	uapServerBase = "apidanalytics_uap_server_base"
+
+	// If caching is used then data scope and developer
+	// info will be maintained in-memory
+	// cache to avoid DB calls for each analytics message
+	useCaching        = "apidanalytics_use_caching"
+	useCachingDefault = true
+)
+
+// keep track of the services that this plugin will use
+var (
+	log      apid.LogService
+	config   apid.ConfigService
+	data     apid.DataService
+	events   apid.EventsService
+	unsafeDB apid.DB
+	dbMux    sync.RWMutex
+
+	localAnalyticsBaseDir      string
+	localAnalyticsTempDir      string
+	localAnalyticsStagingDir   string
+	localAnalyticsFailedDir    string
+	localAnalyticsRecoveredDir string
+)
+
+// apid.RegisterPlugin() is required to be called in init()
+func init() {
+	apid.RegisterPlugin(initPlugin)
+}
+
+func getDB() apid.DB {
+	dbMux.RLock()
+	db := unsafeDB
+	dbMux.RUnlock()
+	return db
+}
+
+func setDB(db apid.DB) {
+	dbMux.Lock()
+	unsafeDB = db
+	dbMux.Unlock()
+}
+
+// initPlugin will be called by apid to initialize
+func initPlugin(services apid.Services) (apid.PluginData, error) {
+	// set a logger that is annotated for this plugin
+	log = services.Log().ForModule("apidAnalytics")
+	log.Debug("start init for apidAnalytics plugin")
+
+	data = services.Data()
+	events = services.Events()
+	events.Listen("ApigeeSync", &handler{})
+
+	// set configuration
+	err := setConfig(services)
+	if err != nil {
+		return pluginData, err
+	}
+
+	for _, key := range []string{uapServerBase} {
+		if !config.IsSet(key) {
+			return pluginData,
+				fmt.Errorf("Missing required config value: %s", key)
+		}
+	}
+
+	// Create directories for managing buffering and upload to UAP stages
+	directories := []string{localAnalyticsBaseDir,
+		localAnalyticsTempDir,
+		localAnalyticsStagingDir,
+		localAnalyticsFailedDir,
+		localAnalyticsRecoveredDir}
+	err = createDirectories(directories)
+
+	if err != nil {
+		return pluginData, fmt.Errorf("Cannot create "+
+			"required local directories: %v ", err)
+	}
+
+	// Initialize one time crash recovery to be performed by the plugin on start up
+	initCrashRecovery()
+
+	// Initialize upload manager to watch the staging directory and
+	// upload files to UAP as they are ready
+	initUploadManager()
+
+	// Initialize buffer manager to watch the internalBuffer channel
+	// for new messages and dump them to files
+	initBufferingManager()
+
+	// Initialize API's and expose them
+	initAPI(services)
+	log.Debug("end init for apidAnalytics plugin")
+	return pluginData, nil
+}
+
+func setConfig(services apid.Services) error {
+	config = services.Config()
+
+	// set plugin config defaults
+	config.SetDefault(configAnalyticsBasePath, analyticsBasePathDefault)
+	config.SetDefault(configAnalyticsDataPath, analyticsDataPathDefault)
+
+	if !config.IsSet("local_storage_path") {
+		return fmt.Errorf("Missing required config value: local_storage_path")
+	}
+
+	// set local directory paths that will be used to buffer analytics data on disk
+	localAnalyticsBaseDir = filepath.Join(config.GetString("local_storage_path"),
+		config.GetString(configAnalyticsDataPath))
+	localAnalyticsTempDir = filepath.Join(localAnalyticsBaseDir, "tmp")
+	localAnalyticsStagingDir = filepath.Join(localAnalyticsBaseDir, "staging")
+	localAnalyticsFailedDir = filepath.Join(localAnalyticsBaseDir, "failed")
+	localAnalyticsRecoveredDir = filepath.Join(localAnalyticsBaseDir, "recovered")
+
+	// set default config for collection interval
+	config.SetDefault(analyticsCollectionInterval, analyticsCollectionIntervalDefault)
+
+	// set default config for useCaching
+	config.SetDefault(useCaching, useCachingDefault)
+
+	// set default config for upload interval
+	config.SetDefault(analyticsUploadInterval, analyticsUploadIntervalDefault)
+
+	// set default config for internal buffer size
+	config.SetDefault(analyticsBufferChannelSize, analyticsBufferChannelSizeDefault)
+
+	return nil
+}
+
+// create all missing directories if required
+func createDirectories(directories []string) error {
+	for _, path := range directories {
+		if _, err := os.Stat(path); os.IsNotExist(err) {
+			error := os.Mkdir(path, os.ModePerm)
+			if error != nil {
+				return error
+			}
+			log.Infof("created directory for analytics "+
+				"data collection: %s", path)
+		}
+	}
+	return nil
+}
diff --git a/listener.go b/listener.go
new file mode 100644
index 0000000..4c57da8
--- /dev/null
+++ b/listener.go
@@ -0,0 +1,120 @@
+package apidAnalytics
+
+import (
+	"github.com/30x/apid"
+	"github.com/apigee-labs/transicator/common"
+)
+
+type handler struct{}
+
+func (h *handler) String() string {
+	return "apigeeAnalytics"
+}
+
+func (h *handler) Handle(e apid.Event) {
+	snapData, ok := e.(*common.Snapshot)
+	if ok {
+		processSnapshot(snapData)
+	} else {
+		changeSet, ok := e.(*common.ChangeList)
+		if ok {
+			processChange(changeSet)
+		} else {
+			log.Errorf("Received Invalid event. Ignoring. %v", e)
+		}
+	}
+	return
+}
+
+func processSnapshot(snapshot *common.Snapshot) {
+	log.Debugf("Snapshot received. Switching to"+
+		" DB version: %s", snapshot.SnapshotInfo)
+
+	db, err := data.DBVersion(snapshot.SnapshotInfo)
+	if err != nil {
+		log.Panicf("Unable to access database: %v", err)
+	}
+	setDB(db)
+
+	if config.GetBool(useCaching) {
+		err = createTenantCache()
+		if err != nil {
+			log.Error(err)
+		} else {
+			log.Debug("Created a local cache" +
+				" for datasope information")
+		}
+		err = createDeveloperInfoCache()
+		if err != nil {
+			log.Error(err)
+		} else {
+			log.Debug("Created a local cache for developer information")
+		}
+	} else {
+		log.Info("Will not be caching any developer info " +
+			"and make a DB call for every analytics msg")
+	}
+	return
+}
+
+func processChange(changes *common.ChangeList) {
+	if config.GetBool(useCaching) {
+		log.Debugf("apigeeSyncEvent: %d changes", len(changes.Changes))
+		var rows []common.Row
+
+		refreshDevInfoNeeded := false
+		for _, payload := range changes.Changes {
+			rows = nil
+			switch payload.Table {
+			case "edgex.data_scope":
+				switch payload.Operation {
+				case common.Insert, common.Update:
+					rows = append(rows, payload.NewRow)
+					// Lock before writing to the
+					// map as it has multiple readers
+					tenantCachelock.Lock()
+					defer tenantCachelock.Unlock()
+					for _, ele := range rows {
+						var scopeuuid, tenantid string
+						var org, env string
+						ele.Get("id", &scopeuuid)
+						ele.Get("scope", &tenantid)
+						ele.Get("org", &org)
+						ele.Get("env", &env)
+						tenantCache[scopeuuid] = tenant{
+							Org:      org,
+							Env:      env,
+							TenantId: tenantid}
+						log.Debugf("Refreshed local "+
+							"tenantCache. Added "+
+							"scope: "+"%s", scopeuuid)
+					}
+				case common.Delete:
+					rows = append(rows, payload.OldRow)
+					// Lock before writing to the map
+					// as it has multiple readers
+					tenantCachelock.Lock()
+					defer tenantCachelock.Unlock()
+					for _, ele := range rows {
+						var scopeuuid string
+						ele.Get("id", &scopeuuid)
+						delete(tenantCache, scopeuuid)
+						log.Debugf("Refreshed local"+
+							" tenantCache. Deleted"+
+							" scope: %s", scopeuuid)
+					}
+				}
+			case "kms.developer", "kms.app", "kms.api_product",
+				"kms.app_credential_apiproduct_mapper":
+				// any change in any of the above tables
+				// should result in cache refresh
+				refreshDevInfoNeeded = true
+			}
+		}
+		// Refresh cache once for all set of changes
+		if refreshDevInfoNeeded {
+			createDeveloperInfoCache()
+			log.Debug("Refresh local developerInfoCache")
+		}
+	}
+}
diff --git a/pluginData.go b/pluginData.go
new file mode 100644
index 0000000..af0faa6
--- /dev/null
+++ b/pluginData.go
@@ -0,0 +1,11 @@
+package apidAnalytics
+
+import "github.com/30x/apid"
+
+var pluginData = apid.PluginData{
+	Name:    "apidAnalytics",
+	Version: "0.0.1",
+	ExtraData: map[string]interface{}{
+		"schemaVersion": "0.0.1",
+	},
+}
diff --git a/upload_manager.go b/upload_manager.go
new file mode 100644
index 0000000..a95ce87
--- /dev/null
+++ b/upload_manager.go
@@ -0,0 +1,111 @@
+package apidAnalytics
+
+import (
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"time"
+)
+
+const (
+	maxRetries              = 3
+	retryFailedDirBatchSize = 10
+)
+
+// Each file upload is retried maxRetries times before
+// moving it to failed directory
+var retriesMap map[string]int
+
+//TODO:  make sure that this instance gets initialized only once
+// since we dont want multiple upload manager tickers running
+func initUploadManager() {
+
+	retriesMap = make(map[string]int)
+
+	go func() {
+		// Periodically check the staging directory to check
+		// if any folders are ready to be uploaded to S3
+		ticker := time.NewTicker(time.Second *
+			config.GetDuration(analyticsUploadInterval))
+		log.Debugf("Intialized upload manager to check for staging directory")
+		// Ticker will keep running till go routine is running
+		// i.e. till application is running
+		defer ticker.Stop()
+
+		for range ticker.C {
+			files, err := ioutil.ReadDir(localAnalyticsStagingDir)
+
+			if err != nil {
+				log.Errorf("Cannot read directory: "+
+					"%s", localAnalyticsStagingDir)
+			}
+
+			uploadedDirCnt := 0
+			for _, file := range files {
+				if file.IsDir() {
+					status := uploadDir(file)
+					handleUploadDirStatus(file, status)
+					if status {
+						uploadedDirCnt++
+					}
+				}
+			}
+			if uploadedDirCnt > 0 {
+				// After a successful upload, retry the
+				// folders in failed directory as they might have
+				// failed due to intermitent S3/GCS issue
+				retryFailedUploads()
+			}
+		}
+	}()
+}
+
+func handleUploadDirStatus(dir os.FileInfo, status bool) {
+	completePath := filepath.Join(localAnalyticsStagingDir, dir.Name())
+	// If upload is successful then delete files
+	// and remove bucket from retry map
+	if status {
+		os.RemoveAll(completePath)
+		log.Debugf("deleted directory after "+
+			"successful upload: %s", dir.Name())
+		// remove key if exists from retry map after a successful upload
+		delete(retriesMap, dir.Name())
+	} else {
+		retriesMap[dir.Name()] = retriesMap[dir.Name()] + 1
+		if retriesMap[dir.Name()] >= maxRetries {
+			log.Errorf("Max Retires exceeded for folder: %s", completePath)
+			failedDirPath := filepath.Join(localAnalyticsFailedDir, dir.Name())
+			err := os.Rename(completePath, failedDirPath)
+			if err != nil {
+				log.Errorf("Cannot move directory '%s'"+
+					" from staging to failed folder", dir.Name())
+			}
+			// remove key from retry map once it reaches allowed max failed attempts
+			delete(retriesMap, dir.Name())
+		}
+	}
+}
+
+func retryFailedUploads() {
+	failedDirs, err := ioutil.ReadDir(localAnalyticsFailedDir)
+
+	if err != nil {
+		log.Errorf("Cannot read directory: %s", localAnalyticsFailedDir)
+	}
+
+	cnt := 0
+	for _, dir := range failedDirs {
+		// We rety failed folder in batches to not overload the upload thread
+		if cnt < retryFailedDirBatchSize {
+			failedPath := filepath.Join(localAnalyticsFailedDir, dir.Name())
+			newStagingPath := filepath.Join(localAnalyticsStagingDir, dir.Name())
+			err := os.Rename(failedPath, newStagingPath)
+			if err != nil {
+				log.Errorf("Cannot move directory '%s'"+
+					" from failed to staging folder", dir.Name())
+			}
+		} else {
+			break
+		}
+	}
+}
diff --git a/uploader.go b/uploader.go
new file mode 100644
index 0000000..fb8a267
--- /dev/null
+++ b/uploader.go
@@ -0,0 +1,155 @@
+package apidAnalytics
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+)
+
+const timestampLayout = "20060102150405" // same as yyyyMMddHHmmss
+
+var token string
+
+var client *http.Client = &http.Client{
+	//set default timeout of 60 seconds while connecting to s3/GCS
+	Timeout: time.Duration(60 * time.Second),
+}
+
+func addHeaders(req *http.Request) {
+	token = config.GetString("apigeesync_bearer_token")
+	req.Header.Add("Authorization", "Bearer "+token)
+}
+
+func uploadDir(dir os.FileInfo) bool {
+	// Eg. org~env~20160101224500
+	tenant, timestamp := splitDirName(dir.Name())
+	//date=2016-01-01/time=22-45
+	dateTimePartition := getDateFromDirTimestamp(timestamp)
+
+	completePath := filepath.Join(localAnalyticsStagingDir, dir.Name())
+	files, _ := ioutil.ReadDir(completePath)
+
+	status := true
+	var error error
+	for _, file := range files {
+		completeFilePath := filepath.Join(completePath, file.Name())
+		relativeFilePath := dateTimePartition + "/" + file.Name()
+		status, error = uploadFile(tenant, relativeFilePath, completeFilePath)
+		if error != nil {
+			log.Errorf("Upload failed due to: %v", error)
+			break
+		} else {
+			os.Remove(completeFilePath)
+			log.Debugf("Deleted file '%s' after "+
+				"successful upload", file.Name())
+		}
+	}
+	return status
+}
+
+func uploadFile(tenant, relativeFilePath, completeFilePath string) (bool, error) {
+	signedUrl, err := getSignedUrl(tenant, relativeFilePath)
+	if err != nil {
+		return false, err
+	} else {
+		return uploadFileToDatastore(completeFilePath, signedUrl)
+	}
+}
+
+func getSignedUrl(tenant, relativeFilePath string) (string, error) {
+	uapCollectionUrl := config.GetString(uapServerBase) + "/analytics"
+
+	req, err := http.NewRequest("GET", uapCollectionUrl, nil)
+	if err != nil {
+		return "", err
+	}
+
+	q := req.URL.Query()
+
+	// eg. edgexfeb1~test
+	q.Add("tenant", tenant)
+	// eg. date=2017-01-30/time=16-32/1069_20170130163200.20170130163400_218e3d99-efaf-4a7b-b3f2-5e4b00c023b7_writer_0.txt.gz
+	q.Add("relative_file_path", relativeFilePath)
+	q.Add("file_content_type", "application/x-gzip")
+	req.URL.RawQuery = q.Encode()
+
+	// Add Bearer Token to each request
+	addHeaders(req)
+	resp, err := client.Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+
+	respBody, _ := ioutil.ReadAll(resp.Body)
+	if resp.StatusCode == 200 {
+		var body map[string]interface{}
+		json.Unmarshal(respBody, &body)
+		signedURL := body["url"]
+		return signedURL.(string), nil
+	} else {
+		return "", fmt.Errorf("Error while getting "+
+			"signed URL '%v'", resp.Status)
+	}
+}
+
+func uploadFileToDatastore(completeFilePath, signedUrl string) (bool, error) {
+	// open gzip file that needs to be uploaded
+	file, err := os.Open(completeFilePath)
+	if err != nil {
+		return false, err
+	}
+	defer file.Close()
+
+	req, err := http.NewRequest("PUT", signedUrl, file)
+	if err != nil {
+		return false, fmt.Errorf("Parsing URL failed '%v'", err)
+	}
+
+	req.Header.Set("Expect", "100-continue")
+	req.Header.Set("Content-Type", "application/x-gzip")
+
+	fileStats, err := file.Stat()
+	if err != nil {
+		return false, fmt.Errorf("Could not get content length for "+
+			"file '%v'", err)
+	}
+	req.ContentLength = fileStats.Size()
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return false, err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode == 200 {
+		return true, nil
+	} else {
+		return false, fmt.Errorf("Final Datastore (S3/GCS)returned "+
+			"Error '%v'", resp.Status)
+	}
+}
+
+// Extract tenant and timestamp from directory Name
+func splitDirName(dirName string) (string, string) {
+	s := strings.Split(dirName, "~")
+	tenant := s[0] + "~" + s[1]
+	timestamp := s[2]
+	return tenant, timestamp
+}
+
+// files are uploaded to S3 under specific date time partition and that
+// key needs to be generated from the plugin
+// eg. <...prefix generated by uap collection service...>/date=2016-01-02/time=15-45/filename.txt.gz
+func getDateFromDirTimestamp(timestamp string) string {
+	dateTime, _ := time.Parse(timestampLayout, timestamp)
+	date := dateTime.Format("2006-01-02") // same as YYYY-MM-dd
+	time := dateTime.Format("15-04")      // same as HH-mm
+	dateHourTS := "date=" + date + "/time=" + time
+	return dateHourTS
+}