[XAPID-854] Listen to Apid shutdown, register callback and do a clean shutdown by closing open files

buffering_manager.go

diff --git a/buffering_manager.go b/buffering_manager.go
index 7e28d4b..6fb156d 100644
--- a/buffering_manager.go
+++ b/buffering_manager.go
@@ -9,6 +9,7 @@
 	"os"
 	"path/filepath"
 	"time"
+	"sync"
 )
 
 const fileExtension = ".txt.gz"
@@ -24,7 +25,12 @@
 // Map from timestampt to bucket
 var bucketMap map[int64]bucket
 
+// RW lock for bucketMap  since the cache can be
+// read while its being written to and vice versa
+var bucketMaplock = sync.RWMutex{}
+
 type bucket struct {
+	keyTS int64
 	DirName string
 	// We need file handle and writer to close the file
 	FileWriter fileWriter
@@ -41,7 +47,11 @@
 	internalBuffer = make(chan axRecords,
 		config.GetInt(analyticsBufferChannelSize))
 	closeBucketEvent = make(chan bucket)
+
+	bucketMaplock.Lock()
 	bucketMap = make(map[int64]bucket)
+	bucketMaplock.Unlock()
+
 
 	// Keep polling the internal buffer for new messages
 	go func() {
@@ -71,9 +81,14 @@
 			// 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)
+				log.Errorf("Cannot move directory '%s' from" +
+					" tmp to staging folder due to '%s", bucket.DirName, err)
 			}
+
+			// Remove bucket from bucket map once its closed successfully
+			bucketMaplock.Lock()
+			delete(bucketMap, bucket.keyTS)
+			bucketMaplock.Unlock()
 		}
 	}()
 }
@@ -92,9 +107,13 @@
 	// 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]
+
+	bucketMaplock.RLock()
+	b, exists := bucketMap[ts]
+	bucketMaplock.RUnlock()
+
 	if exists {
-		return bucketMap[ts], nil
+		return b, nil
 	} else {
 		timestamp := time.Unix(ts, 0).Format(timestampLayout)
 
@@ -123,8 +142,11 @@
 			return bucket{}, err
 		}
 
-		newBucket := bucket{DirName: dirName, FileWriter: fw}
+		newBucket := bucket{keyTS: ts, DirName: dirName, FileWriter: fw}
+
+		bucketMaplock.Lock()
 		bucketMap[ts] = newBucket
+		bucketMaplock.Unlock()
 
 		//Send event to close directory after endTime + 5
 		// seconds to make sure all buffers are flushed to file

buffering_manager_test.go

diff --git a/buffering_manager_test.go b/buffering_manager_test.go
index fa4ad85..c3c747e 100644
--- a/buffering_manager_test.go
+++ b/buffering_manager_test.go
@@ -118,7 +118,7 @@
 			fw, e := createGzipFile(completeFilePath)
 			Expect(e).ShouldNot(HaveOccurred())
 
-			bucket := bucket{DirName: dirName, FileWriter: fw}
+			bucket := bucket{keyTS: 112312, DirName: dirName, FileWriter: fw}
 			closeBucketEvent <- bucket
 
 			// wait for it to close dir and move to staging

init.go

diff --git a/init.go b/init.go
index 0706e57..386aa1e 100644
--- a/init.go
+++ b/init.go
@@ -143,6 +143,15 @@
 		}()
 	}
 
+	// Create a listener for shutdown event and register callback
+	h := func(event apid.Event) {
+		log.Infof("Received ApidShutdown event. %v", event)
+		shutdownPlugin();
+		return
+	}
+	log.Infof("registered listener for shutdown event")
+	events.ListenOnceFunc(apid.ShutdownEventSelector, h)
+
 	// Initialize API's and expose them
 	initAPI(services)
 	log.Debug("end init for apidAnalytics plugin")
@@ -200,3 +209,29 @@
 	}
 	return nil
 }
+
+func shutdownPlugin() {
+	log.Info("Shutting down apidAnalytics plugin")
+
+	// Close all open files and move directories in tmp to staging.
+	bucketMaplock.RLock()
+	defer bucketMaplock.RUnlock()
+	for _, bucket := range bucketMap {
+		log.Infof("closing bucket '%s' as a part of shutdown", bucket.DirName)
+		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 due to '%s", bucket.DirName, err)
+		}
+
+		bucketMaplock.Lock()
+		delete(bucketMap, bucket.keyTS)
+		bucketMaplock.Unlock()
+	}
+}
\ No newline at end of file

upload_manager.go

diff --git a/upload_manager.go b/upload_manager.go
index 4deba03..7373daf 100644
--- a/upload_manager.go
+++ b/upload_manager.go
@@ -47,6 +47,8 @@
 					handleUploadDirStatus(file, status)
 					if status {
 						uploadedDirCnt++
+						log.Infof("Successfully uploaded: %s",
+							file.Name())
 					}
 				}
 			}