Migrating from consul to mongodb for backend

Migrating to mongodb from consul.
The main reason being the value size limitation of 512kb in consul.
See https://jira.onap.org/browse/MULTICLOUD-426 for details.
This requires a little bit of hierarchy management
and data management.
We are no longer converting structs to json encoded
strings. The underlying db supports structs without any modifications.
Also, since Mongo has the concept of collections, each submodule can
use its own collection for storage as needed.
Definition uses a collection called rbdef right now.
P10: Enabling unit tests for mongo.go. This requires the usage
     of aliased functions.
P11: Expanded unit tests for all functions in mongo.go
P12: Refactored parameter validation.
     Removed TestHealthCheck as we are not mocking any of the
       db commands right now
     Checking return value of read with an expected value
P13: Adding back consul support.
     Fixing functional test
     Full consul implementation check and modifications is
     being tracked by MULTICLOUD-427
P15: Fix ReadAll unit test and corresponding code
     ReadAll now returns error when no objects are found

Issue-ID: MULTICLOUD-426
Change-Id: I42d239b324025fc4ef4e561790aceeff794001ef
Signed-off-by: Kiran Kamineni <kiran.k.kamineni@intel.com>

7 files changed, 965 insertions, 62 deletions

diff --git a/src/k8splugin/db/consul.go b/src/k8splugin/db/consul.go
index d7507242..a61a4c10 100644
--- a/src/k8splugin/db/consul.go
+++ b/src/k8splugin/db/consul.go
@@ -54,50 +54,64 @@ func NewConsulStore(store ConsulKVStore) (Store, error) {
 
 // HealthCheck verifies if the database is up and running
 func (c *ConsulStore) HealthCheck() error {
-	_, err := c.Read("test")
+	_, err := c.Read("test", "test", "test")
 	if err != nil {
 		return pkgerrors.New("[ERROR] Cannot talk to Datastore. Check if it is running/reachable.")
 	}
 	return nil
 }
 
+// Unmarshal implements any unmarshaling that is needed when using consul
+func (c *ConsulStore) Unmarshal(inp []byte, out interface{}) error {
+	return nil
+}
+
 // Create is used to create a DB entry
-func (c *ConsulStore) Create(key, value string) error {
+func (c *ConsulStore) Create(root, key, tag string, data interface{}) error {
+
+	value, err := Serialize(data)
+	if err != nil {
+		return pkgerrors.Wrap(err, "Serializing input data")
+	}
+
 	p := &api.KVPair{
 		Key:   key,
 		Value: []byte(value),
 	}
-	_, err := c.client.Put(p, nil)
+	_, err = c.client.Put(p, nil)
 	return err
 }
 
 // Read method returns the internalID for a particular externalID
-func (c *ConsulStore) Read(key string) (string, error) {
+func (c *ConsulStore) Read(root, key, tag string) ([]byte, error) {
+	key = root + "/" + key + "/" + tag
 	pair, _, err := c.client.Get(key, nil)
 	if err != nil {
-		return "", err
+		return nil, err
 	}
 	if pair == nil {
-		return "", nil
+		return nil, nil
 	}
-	return string(pair.Value), nil
+	return pair.Value, nil
 }
 
 // Delete method removes an internalID from the Database
-func (c *ConsulStore) Delete(key string) error {
+func (c *ConsulStore) Delete(root, key, tag string) error {
 	_, err := c.client.Delete(key, nil)
 	return err
 }
 
 // ReadAll is used to get all ExternalIDs in a namespace
-func (c *ConsulStore) ReadAll(prefix string) ([]string, error) {
-	pairs, _, err := c.client.List(prefix, nil)
+func (c *ConsulStore) ReadAll(root, tag string) (map[string][]byte, error) {
+	pairs, _, err := c.client.List(root, nil)
 	if err != nil {
 		return nil, err
 	}
-	var result []string
+
+	//TODO: Filter results by tag and return it
+	result := make(map[string][]byte)
 	for _, keypair := range pairs {
-		result = append(result, keypair.Key)
+		result[keypair.Key] = keypair.Value
 	}
 
 	return result, nil
diff --git a/src/k8splugin/db/consul_test.go b/src/k8splugin/db/consul_test.go
index ede1a5e9..754112ad 100644
--- a/src/k8splugin/db/consul_test.go
+++ b/src/k8splugin/db/consul_test.go
@@ -107,12 +107,14 @@ func TestConsulCreate(t *testing.T) {
 	}{
 		{
 			label: "Sucessful register a record to Consul Database",
-			input: map[string]string{"key": "test-key", "value": "test-value"},
-			mock:  &mockConsulKVStore{},
+			input: map[string]string{"root": "rbinst", "key": "test-key",
+				"tag": "data", "value": "test-value"},
+			mock: &mockConsulKVStore{},
 		},
 		{
 			label: "Fail to create a new record in Consul Database",
-			input: map[string]string{"key": "test-key", "value": "test-value"},
+			input: map[string]string{"root": "rbinst", "key": "test-key",
+				"tag": "data", "value": "test-value"},
 			mock: &mockConsulKVStore{
 				Err: pkgerrors.New("DB error"),
 			},
@@ -123,7 +125,8 @@ func TestConsulCreate(t *testing.T) {
 	for _, testCase := range testCases {
 		t.Run(testCase.label, func(t *testing.T) {
 			client, _ := NewConsulStore(testCase.mock)
-			err := client.Create(testCase.input["key"], testCase.input["value"])
+			err := client.Create(testCase.input["root"], testCase.input["key"],
+				testCase.input["tag"], testCase.input["value"])
 			if err != nil {
 				if testCase.expectedError == "" {
 					t.Fatalf("Create method return an un-expected (%s)", err)
@@ -139,18 +142,19 @@ func TestConsulCreate(t *testing.T) {
 func TestConsulRead(t *testing.T) {
 	testCases := []struct {
 		label          string
-		input          string
+		input          map[string]string
 		mock           *mockConsulKVStore
 		expectedError  string
 		expectedResult string
 	}{
 		{
 			label: "Sucessful retrieve a record from Consul Database",
-			input: "test",
+			input: map[string]string{"root": "rbinst", "key": "test",
+				"tag": "data"},
 			mock: &mockConsulKVStore{
 				Items: api.KVPairs{
 					&api.KVPair{
-						Key:   "test",
+						Key:   "rbinst/test/data",
 						Value: []byte("test-value"),
 					},
 				},
@@ -159,12 +163,14 @@ func TestConsulRead(t *testing.T) {
 		},
 		{
 			label: "Fail retrieve a non-existing record from Consul Database",
-			input: "test",
-			mock:  &mockConsulKVStore{},
+			input: map[string]string{"root": "rbinst", "key": "test-key",
+				"tag": "data"},
+			mock: &mockConsulKVStore{},
 		},
 		{
 			label: "Fail retrieve a record from Consul Database",
-			input: "test",
+			input: map[string]string{"root": "rbinst", "key": "test-key",
+				"tag": "data"},
 			mock: &mockConsulKVStore{
 				Err: pkgerrors.New("DB error"),
 			},
@@ -175,7 +181,8 @@ func TestConsulRead(t *testing.T) {
 	for _, testCase := range testCases {
 		t.Run(testCase.label, func(t *testing.T) {
 			client, _ := NewConsulStore(testCase.mock)
-			result, err := client.Read(testCase.input)
+			result, err := client.Read(testCase.input["root"], testCase.input["key"],
+				testCase.input["tag"])
 			if err != nil {
 				if testCase.expectedError == "" {
 					t.Fatalf("Read method return an un-expected (%s)", err)
@@ -187,7 +194,7 @@ func TestConsulRead(t *testing.T) {
 				if testCase.expectedError != "" && testCase.expectedResult == "" {
 					t.Fatalf("Read method was expecting \"%s\" error message", testCase.expectedError)
 				}
-				if !reflect.DeepEqual(testCase.expectedResult, result) {
+				if !reflect.DeepEqual(testCase.expectedResult, string(result)) {
 
 					t.Fatalf("Read method returned: \n%v\n and it was expected: \n%v", result, testCase.expectedResult)
 				}
@@ -199,14 +206,15 @@ func TestConsulRead(t *testing.T) {
 func TestConsulDelete(t *testing.T) {
 	testCases := []struct {
 		label         string
-		input         string
+		input         map[string]string
 		mock          *mockConsulKVStore
 		expectedError string
 	}{
 		{
 			label: "Sucessful delete a record to Consul Database",
-			input: "test",
-			mock:  &mockConsulKVStore{},
+			input: map[string]string{"root": "rbinst", "key": "test-key",
+				"tag": "data"},
+			mock: &mockConsulKVStore{},
 		},
 		{
 			label: "Fail to delete a record in Consul Database",
@@ -220,7 +228,8 @@ func TestConsulDelete(t *testing.T) {
 	for _, testCase := range testCases {
 		t.Run(testCase.label, func(t *testing.T) {
 			client, _ := NewConsulStore(testCase.mock)
-			err := client.Delete(testCase.input)
+			err := client.Delete(testCase.input["root"], testCase.input["key"],
+				testCase.input["tag"])
 			if err != nil {
 				if testCase.expectedError == "" {
 					t.Fatalf("Delete method return an un-expected (%s)", err)
@@ -236,14 +245,15 @@ func TestConsulDelete(t *testing.T) {
 func TestConsulReadAll(t *testing.T) {
 	testCases := []struct {
 		label          string
-		input          string
+		input          map[string]string
 		mock           *mockConsulKVStore
 		expectedError  string
-		expectedResult []string
+		expectedResult map[string][]byte
 	}{
 		{
 			label: "Sucessful retrieve a list from Consul Database",
-			input: "test",
+			input: map[string]string{"root": "rbinst", "key": "test-key",
+				"tag": "data"},
 			mock: &mockConsulKVStore{
 				Items: api.KVPairs{
 					&api.KVPair{
@@ -256,16 +266,20 @@ func TestConsulReadAll(t *testing.T) {
 					},
 				},
 			},
-			expectedResult: []string{"test", "test2"},
+			expectedResult: map[string][]byte{"test": []byte("test-value"),
+				"test2": []byte("test-value2")},
 		},
 		{
 			label: "Sucessful retrieve an empty list from Consul Database",
-			input: "test",
-			mock:  &mockConsulKVStore{},
+			input: map[string]string{"root": "rbinst", "key": "test-key",
+				"tag": "data"},
+			mock:           &mockConsulKVStore{},
+			expectedResult: map[string][]byte{},
 		},
 		{
 			label: "Fail retrieve a record from Consul Database",
-			input: "test",
+			input: map[string]string{"root": "rbinst", "key": "test-key",
+				"tag": "data"},
 			mock: &mockConsulKVStore{
 				Err: pkgerrors.New("DB error"),
 			},
@@ -276,7 +290,8 @@ func TestConsulReadAll(t *testing.T) {
 	for _, testCase := range testCases {
 		t.Run(testCase.label, func(t *testing.T) {
 			client, _ := NewConsulStore(testCase.mock)
-			result, err := client.ReadAll(testCase.input)
+			result, err := client.ReadAll(testCase.input["root"],
+				testCase.input["tag"])
 			if err != nil {
 				if testCase.expectedError == "" {
 					t.Fatalf("ReadAll method return an un-expected (%s)", err)
diff --git a/src/k8splugin/db/mongo.go b/src/k8splugin/db/mongo.go
new file mode 100644
index 00000000..311f044c
--- /dev/null
+++ b/src/k8splugin/db/mongo.go
@@ -0,0 +1,323 @@
+/*
+ * Copyright 2018 Intel Corporation, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package db
+
+import (
+	"github.com/mongodb/mongo-go-driver/bson"
+	"github.com/mongodb/mongo-go-driver/bson/primitive"
+	"github.com/mongodb/mongo-go-driver/mongo"
+	"github.com/mongodb/mongo-go-driver/mongo/options"
+	pkgerrors "github.com/pkg/errors"
+	"golang.org/x/net/context"
+	"log"
+	"os"
+)
+
+// MongoCollection defines the a subset of MongoDB operations
+// Note: This interface is defined mainly for mock testing
+type MongoCollection interface {
+	InsertOne(ctx context.Context, document interface{},
+		opts ...*options.InsertOneOptions) (*mongo.InsertOneResult, error)
+	FindOne(ctx context.Context, filter interface{},
+		opts ...*options.FindOneOptions) *mongo.SingleResult
+	FindOneAndUpdate(ctx context.Context, filter interface{},
+		update interface{}, opts ...*options.FindOneAndUpdateOptions) *mongo.SingleResult
+	DeleteOne(ctx context.Context, filter interface{},
+		opts ...*options.DeleteOptions) (*mongo.DeleteResult, error)
+	Find(ctx context.Context, filter interface{},
+		opts ...*options.FindOptions) (mongo.Cursor, error)
+}
+
+// MongoStore is an implementation of the db.Store interface
+type MongoStore struct {
+	db *mongo.Database
+}
+
+// This exists only for allowing us to mock the collection object
+// for testing purposes
+var getCollection = func(coll string, m *MongoStore) MongoCollection {
+	return m.db.Collection(coll)
+}
+
+// This exists only for allowing us to mock the DecodeBytes function
+// Mainly because we cannot construct a SingleResult struct from our
+// tests. All fields in that struct are private.
+var decodeBytes = func(sr *mongo.SingleResult) (bson.Raw, error) {
+	return sr.DecodeBytes()
+}
+
+// NewMongoStore initializes a Mongo Database with the name provided
+// If a database with that name exists, it will be returned
+func NewMongoStore(name string, store *mongo.Database) (Store, error) {
+	if store == nil {
+		ip := "mongodb://" + os.Getenv("DATABASE_IP") + ":27017"
+		mongoClient, err := mongo.NewClient(ip)
+		if err != nil {
+			return nil, err
+		}
+
+		err = mongoClient.Connect(context.Background())
+		if err != nil {
+			return nil, err
+		}
+		store = mongoClient.Database(name)
+	}
+
+	return &MongoStore{
+		db: store,
+	}, nil
+}
+
+// HealthCheck verifies if the database is up and running
+func (m *MongoStore) HealthCheck() error {
+
+	_, err := decodeBytes(m.db.RunCommand(context.Background(), bson.D{{"serverStatus", 1}}))
+	if err != nil {
+		return pkgerrors.Wrap(err, "Error getting server status")
+	}
+
+	return nil
+}
+
+// validateParams checks to see if any parameters are empty
+func (m *MongoStore) validateParams(args ...string) bool {
+	for _, v := range args {
+		if v == "" {
+			return false
+		}
+	}
+
+	return true
+}
+
+// Create is used to create a DB entry
+func (m *MongoStore) Create(coll, key, tag string, data interface{}) error {
+	if data == nil || !m.validateParams(coll, key, tag) {
+		return pkgerrors.New("No Data to store")
+	}
+
+	c := getCollection(coll, m)
+	ctx := context.Background()
+
+	//Insert the data and then add the objectID to the masterTable
+	res, err := c.InsertOne(ctx, bson.D{
+		{tag, data},
+	})
+	if err != nil {
+		return pkgerrors.Errorf("Error inserting into database: %s", err.Error())
+	}
+
+	//Add objectID of created data to masterKey document
+	//Create masterkey document if it does not exist
+	filter := bson.D{{"key", key}}
+
+	_, err = decodeBytes(
+		c.FindOneAndUpdate(
+			ctx,
+			filter,
+			bson.D{
+				{"$set", bson.D{
+					{tag, res.InsertedID},
+				}},
+			},
+			options.FindOneAndUpdate().SetUpsert(true).SetReturnDocument(options.After)))
+
+	if err != nil {
+		return pkgerrors.Errorf("Error updating master table: %s", err.Error())
+	}
+
+	return nil
+}
+
+// Unmarshal implements an unmarshaler for bson data that
+// is produced from the mongo database
+func (m *MongoStore) Unmarshal(inp []byte, out interface{}) error {
+	err := bson.Unmarshal(inp, out)
+	if err != nil {
+		return pkgerrors.Wrap(err, "Unmarshaling bson")
+	}
+	return nil
+}
+
+// Read method returns the data stored for this key and for this particular tag
+func (m *MongoStore) Read(coll, key, tag string) ([]byte, error) {
+	if !m.validateParams(coll, key, tag) {
+		return nil, pkgerrors.New("Mandatory fields are missing")
+	}
+
+	c := getCollection(coll, m)
+	ctx := context.Background()
+
+	//Get the masterkey document based on given key
+	filter := bson.D{{"key", key}}
+	keydata, err := decodeBytes(c.FindOne(context.Background(), filter))
+	if err != nil {
+		return nil, pkgerrors.Errorf("Error finding master table: %s", err.Error())
+	}
+
+	//Read the tag objectID from document
+	tagoid, ok := keydata.Lookup(tag).ObjectIDOK()
+	if !ok {
+		return nil, pkgerrors.Errorf("Error finding objectID for tag %s", tag)
+	}
+
+	//Use tag objectID to read the data from store
+	filter = bson.D{{"_id", tagoid}}
+	tagdata, err := decodeBytes(c.FindOne(ctx, filter))
+	if err != nil {
+		return nil, pkgerrors.Errorf("Error reading found object: %s", err.Error())
+	}
+
+	//Return the data as a byte array
+	return tagdata.Lookup(tag).Value, nil
+}
+
+// Helper function that deletes an object by its ID
+func (m *MongoStore) deleteObjectByID(coll string, objID primitive.ObjectID) error {
+
+	c := getCollection(coll, m)
+	ctx := context.Background()
+
+	_, err := c.DeleteOne(ctx, bson.D{{"_id", objID}})
+	if err != nil {
+		return pkgerrors.Errorf("Error Deleting from database: %s", err.Error())
+	}
+
+	log.Printf("Deleted Obj with ID %s", objID.String())
+	return nil
+}
+
+// Delete method removes a document from the Database that matches key
+// TODO: delete all referenced docs if tag is empty string
+func (m *MongoStore) Delete(coll, key, tag string) error {
+	if !m.validateParams(coll, key, tag) {
+		return pkgerrors.New("Mandatory fields are missing")
+	}
+
+	c := getCollection(coll, m)
+	ctx := context.Background()
+
+	//Get the masterkey document based on given key
+	filter := bson.D{{"key", key}}
+	//Remove the tag ID entry from masterkey table
+	update := bson.D{
+		{
+			"$unset", bson.D{
+				{tag, ""},
+			},
+		},
+	}
+	keydata, err := decodeBytes(c.FindOneAndUpdate(ctx, filter, update,
+		options.FindOneAndUpdate().SetReturnDocument(options.Before)))
+	if err != nil {
+		return pkgerrors.Errorf("Error decoding master table after update: %s",
+			err.Error())
+	}
+
+	//Read the tag objectID from document
+	elems, err := keydata.Elements()
+	if err != nil {
+		return pkgerrors.Errorf("Error reading elements from database: %s", err.Error())
+	}
+
+	tagoid, ok := keydata.Lookup(tag).ObjectIDOK()
+	if !ok {
+		return pkgerrors.Errorf("Error finding objectID for tag %s", tag)
+	}
+
+	//Use tag objectID to read the data from store
+	err = m.deleteObjectByID(coll, tagoid)
+	if err != nil {
+		return pkgerrors.Errorf("Error deleting from database: %s", err.Error())
+	}
+
+	//Delete master table if no more tags left
+	//_id, key and tag should be elements in before doc
+	//if master table needs to be removed too
+	if len(elems) == 3 {
+		keyid, ok := keydata.Lookup("_id").ObjectIDOK()
+		if !ok {
+			return pkgerrors.Errorf("Error finding objectID for key %s", key)
+		}
+		err = m.deleteObjectByID(coll, keyid)
+		if err != nil {
+			return pkgerrors.Errorf("Error deleting master table from database: %s", err.Error())
+		}
+	}
+
+	return nil
+}
+
+// ReadAll is used to get all documents in db of a particular tag
+func (m *MongoStore) ReadAll(coll, tag string) (map[string][]byte, error) {
+	if !m.validateParams(coll, tag) {
+		return nil, pkgerrors.New("Missing collection or tag name")
+	}
+
+	c := getCollection(coll, m)
+	ctx := context.Background()
+
+	//Get all master tables in this collection
+	filter := bson.D{
+		{"key", bson.D{
+			{"$exists", true},
+		}},
+	}
+	cursor, err := c.Find(ctx, filter)
+	if err != nil {
+		return nil, pkgerrors.Errorf("Error reading from database: %s", err.Error())
+	}
+	defer cursor.Close(ctx)
+
+	//Iterate over all the master tables
+	result := make(map[string][]byte)
+	for cursor.Next(ctx) {
+		d, err := cursor.DecodeBytes()
+		if err != nil {
+			log.Printf("Unable to decode data in Readall: %s", err.Error())
+			continue
+		}
+
+		//Read key of each master table
+		key, ok := d.Lookup("key").StringValueOK()
+		if !ok {
+			log.Printf("Unable to read key string from mastertable %s", err.Error())
+			continue
+		}
+
+		//Get objectID of tag document
+		tid, ok := d.Lookup(tag).ObjectIDOK()
+		if !ok {
+			log.Printf("Did not find tag: %s", tag)
+			continue
+		}
+
+		//Find tag document and unmarshal it into []byte
+		tagData, err := decodeBytes(c.FindOne(ctx, bson.D{{"_id", tid}}))
+		if err != nil {
+			log.Printf("Unable to decode tag data %s", err.Error())
+			continue
+		}
+		result[key] = tagData.Lookup(tag).Value
+	}
+
+	if len(result) == 0 {
+		return result, pkgerrors.Errorf("Did not find any objects with tag: %s", tag)
+	}
+
+	return result, nil
+}
diff --git a/src/k8splugin/db/mongo_test.go b/src/k8splugin/db/mongo_test.go
new file mode 100644
index 00000000..1663e774
--- /dev/null
+++ b/src/k8splugin/db/mongo_test.go
@@ -0,0 +1,530 @@
+// +build unit
+
+/*
+ * Copyright 2018 Intel Corporation, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package db
+
+import (
+	"bytes"
+	"context"
+	"github.com/mongodb/mongo-go-driver/bson"
+	"github.com/mongodb/mongo-go-driver/mongo"
+	"github.com/mongodb/mongo-go-driver/mongo/options"
+	pkgerrors "github.com/pkg/errors"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+// Implements the mongo.Cursor interface
+type mockCursor struct {
+	mongo.Cursor
+	err   error
+	bson  bson.Raw
+	count int
+}
+
+func (mc *mockCursor) Next(ctx context.Context) bool {
+	if mc.count > 0 {
+		mc.count = mc.count - 1
+		return true
+	}
+	return false
+}
+
+func (mc *mockCursor) DecodeBytes() (bson.Raw, error) {
+	return mc.bson, mc.err
+}
+
+func (mc *mockCursor) Close(ctx context.Context) error {
+	return nil
+}
+
+//Implements the functions used currently in mongo.go
+type mockCollection struct {
+	Err     error
+	mCursor mongo.Cursor
+}
+
+func (c *mockCollection) InsertOne(ctx context.Context, document interface{},
+	opts ...*options.InsertOneOptions) (*mongo.InsertOneResult, error) {
+
+	if c.Err != nil {
+		return nil, c.Err
+	}
+
+	return &mongo.InsertOneResult{InsertedID: "_id1234"}, nil
+}
+
+func (c *mockCollection) FindOne(ctx context.Context, filter interface{},
+	opts ...*options.FindOneOptions) *mongo.SingleResult {
+
+	return &mongo.SingleResult{}
+}
+
+func (c *mockCollection) FindOneAndUpdate(ctx context.Context, filter interface{},
+	update interface{}, opts ...*options.FindOneAndUpdateOptions) *mongo.SingleResult {
+
+	return &mongo.SingleResult{}
+}
+
+func (c *mockCollection) DeleteOne(ctx context.Context, filter interface{},
+	opts ...*options.DeleteOptions) (*mongo.DeleteResult, error) {
+
+	return nil, c.Err
+}
+
+func (c *mockCollection) Find(ctx context.Context, filter interface{},
+	opts ...*options.FindOptions) (mongo.Cursor, error) {
+
+	return c.mCursor, c.Err
+}
+
+func TestCreate(t *testing.T) {
+	testCases := []struct {
+		label         string
+		input         map[string]interface{}
+		mockColl      *mockCollection
+		bson          bson.Raw
+		expectedError string
+	}{
+		{
+			label: "Successfull creation of entry",
+			input: map[string]interface{}{
+				"coll": "collname",
+				"key":  "keyvalue",
+				"tag":  "tagName",
+				"data": "Data In String Format",
+			},
+			bson:     bson.Raw{'\x08', '\x00', '\x00', '\x00', '\x0A', 'x', '\x00', '\x00'},
+			mockColl: &mockCollection{},
+		},
+		{
+			label: "UnSuccessfull creation of entry",
+			input: map[string]interface{}{
+				"coll": "collname",
+				"key":  "keyvalue",
+				"tag":  "tagName",
+				"data": "Data In String Format",
+			},
+			mockColl: &mockCollection{
+				Err: pkgerrors.New("DB Error"),
+			},
+			expectedError: "DB Error",
+		},
+		{
+			label: "Missing input fields",
+			input: map[string]interface{}{
+				"coll": "",
+				"key":  "",
+				"tag":  "",
+				"data": "",
+			},
+			expectedError: "No Data to store",
+			mockColl:      &mockCollection{},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.label, func(t *testing.T) {
+			m, _ := NewMongoStore("name", &mongo.Database{})
+			// Override the getCollection function with our mocked version
+			getCollection = func(coll string, m *MongoStore) MongoCollection {
+				return testCase.mockColl
+			}
+
+			decodeBytes = func(sr *mongo.SingleResult) (bson.Raw, error) {
+				return testCase.bson, testCase.mockColl.Err
+			}
+
+			err := m.Create(testCase.input["coll"].(string), testCase.input["key"].(string),
+				testCase.input["tag"].(string), testCase.input["data"])
+			if err != nil {
+				if testCase.expectedError == "" {
+					t.Fatalf("Create method returned an un-expected (%s)", err)
+				}
+				if !strings.Contains(string(err.Error()), testCase.expectedError) {
+					t.Fatalf("Create method returned an error (%s)", err)
+				}
+			}
+		})
+	}
+}
+
+func TestRead(t *testing.T) {
+	testCases := []struct {
+		label         string
+		input         map[string]interface{}
+		mockColl      *mockCollection
+		bson          bson.Raw
+		expectedError string
+		expected      []byte
+	}{
+		{
+			label: "Successfull Read of entry",
+			input: map[string]interface{}{
+				"coll": "collname",
+				"key":  "keyvalue",
+				"tag":  "metadata",
+			},
+			// Binary form of
+			// {
+			//	"_id" : ObjectId("5c115156777ff85654248ae1"),
+			//  "key" : "b82c4bb1-09ff-6093-4d58-8327b94e1e20",
+			//  "metadata" : ObjectId("5c115156c9755047e318bbfd")
+			// }
+			bson: bson.Raw{
+				'\x5a', '\x00', '\x00', '\x00', '\x07', '\x5f', '\x69', '\x64',
+				'\x00', '\x5c', '\x11', '\x51', '\x56', '\x77', '\x7f', '\xf8',
+				'\x56', '\x54', '\x24', '\x8a', '\xe1', '\x02', '\x6b', '\x65',
+				'\x79', '\x00', '\x25', '\x00', '\x00', '\x00', '\x62', '\x38',
+				'\x32', '\x63', '\x34', '\x62', '\x62', '\x31', '\x2d', '\x30',
+				'\x39', '\x66', '\x66', '\x2d', '\x36', '\x30', '\x39', '\x33',
+				'\x2d', '\x34', '\x64', '\x35', '\x38', '\x2d', '\x38', '\x33',
+				'\x32', '\x37', '\x62', '\x39', '\x34', '\x65', '\x31', '\x65',
+				'\x32', '\x30', '\x00', '\x07', '\x6d', '\x65', '\x74', '\x61',
+				'\x64', '\x61', '\x74', '\x61', '\x00', '\x5c', '\x11', '\x51',
+				'\x56', '\xc9', '\x75', '\x50', '\x47', '\xe3', '\x18', '\xbb',
+				'\xfd', '\x00',
+			},
+			mockColl: &mockCollection{},
+			// This is not the document because we are mocking decodeBytes
+			expected: []byte{92, 17, 81, 86, 201, 117, 80, 71, 227, 24, 187, 253},
+		},
+		{
+			label: "UnSuccessfull Read of entry: object not found",
+			input: map[string]interface{}{
+				"coll": "collname",
+				"key":  "keyvalue",
+				"tag":  "badtag",
+			},
+			// Binary form of
+			// {
+			//	"_id" : ObjectId("5c115156777ff85654248ae1"),
+			//  "key" : "b82c4bb1-09ff-6093-4d58-8327b94e1e20",
+			//  "metadata" : ObjectId("5c115156c9755047e318bbfd")
+			// }
+			bson: bson.Raw{
+				'\x5a', '\x00', '\x00', '\x00', '\x07', '\x5f', '\x69', '\x64',
+				'\x00', '\x5c', '\x11', '\x51', '\x56', '\x77', '\x7f', '\xf8',
+				'\x56', '\x54', '\x24', '\x8a', '\xe1', '\x02', '\x6b', '\x65',
+				'\x79', '\x00', '\x25', '\x00', '\x00', '\x00', '\x62', '\x38',
+				'\x32', '\x63', '\x34', '\x62', '\x62', '\x31', '\x2d', '\x30',
+				'\x39', '\x66', '\x66', '\x2d', '\x36', '\x30', '\x39', '\x33',
+				'\x2d', '\x34', '\x64', '\x35', '\x38', '\x2d', '\x38', '\x33',
+				'\x32', '\x37', '\x62', '\x39', '\x34', '\x65', '\x31', '\x65',
+				'\x32', '\x30', '\x00', '\x07', '\x6d', '\x65', '\x74', '\x61',
+				'\x64', '\x61', '\x74', '\x61', '\x00', '\x5c', '\x11', '\x51',
+				'\x56', '\xc9', '\x75', '\x50', '\x47', '\xe3', '\x18', '\xbb',
+				'\xfd', '\x00',
+			},
+			mockColl:      &mockCollection{},
+			expectedError: "Error finding objectID",
+		},
+		{
+			label: "UnSuccessfull Read of entry",
+			input: map[string]interface{}{
+				"coll": "collname",
+				"key":  "keyvalue",
+				"tag":  "tagName",
+			},
+			mockColl: &mockCollection{
+				Err: pkgerrors.New("DB Error"),
+			},
+			expectedError: "DB Error",
+		},
+		{
+			label: "Missing input fields",
+			input: map[string]interface{}{
+				"coll": "",
+				"key":  "",
+				"tag":  "",
+			},
+			expectedError: "Mandatory fields are missing",
+			mockColl:      &mockCollection{},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.label, func(t *testing.T) {
+			m, _ := NewMongoStore("name", &mongo.Database{})
+			// Override the getCollection function with our mocked version
+			getCollection = func(coll string, m *MongoStore) MongoCollection {
+				return testCase.mockColl
+			}
+
+			decodeBytes = func(sr *mongo.SingleResult) (bson.Raw, error) {
+				return testCase.bson, testCase.mockColl.Err
+			}
+			got, err := m.Read(testCase.input["coll"].(string), testCase.input["key"].(string),
+				testCase.input["tag"].(string))
+			if err != nil {
+				if testCase.expectedError == "" {
+					t.Fatalf("Read method returned an un-expected (%s)", err)
+				}
+				if !strings.Contains(string(err.Error()), testCase.expectedError) {
+					t.Fatalf("Read method returned an error (%s)", err)
+				}
+			} else {
+				if bytes.Compare(got, testCase.expected) != 0 {
+					t.Fatalf("Read returned unexpected data: %s, expected: %s",
+						string(got), testCase.expected)
+				}
+			}
+		})
+	}
+}
+
+func TestDelete(t *testing.T) {
+	testCases := []struct {
+		label         string
+		input         map[string]interface{}
+		mockColl      *mockCollection
+		bson          bson.Raw
+		expectedError string
+	}{
+		{
+			label: "Successfull Delete of entry",
+			input: map[string]interface{}{
+				"coll": "collname",
+				"key":  "keyvalue",
+				"tag":  "metadata",
+			},
+			// Binary form of
+			// {
+			//	"_id" : ObjectId("5c115156777ff85654248ae1"),
+			//  "key" : "b82c4bb1-09ff-6093-4d58-8327b94e1e20",
+			//  "metadata" : ObjectId("5c115156c9755047e318bbfd")
+			// }
+			bson: bson.Raw{
+				'\x5a', '\x00', '\x00', '\x00', '\x07', '\x5f', '\x69', '\x64',
+				'\x00', '\x5c', '\x11', '\x51', '\x56', '\x77', '\x7f', '\xf8',
+				'\x56', '\x54', '\x24', '\x8a', '\xe1', '\x02', '\x6b', '\x65',
+				'\x79', '\x00', '\x25', '\x00', '\x00', '\x00', '\x62', '\x38',
+				'\x32', '\x63', '\x34', '\x62', '\x62', '\x31', '\x2d', '\x30',
+				'\x39', '\x66', '\x66', '\x2d', '\x36', '\x30', '\x39', '\x33',
+				'\x2d', '\x34', '\x64', '\x35', '\x38', '\x2d', '\x38', '\x33',
+				'\x32', '\x37', '\x62', '\x39', '\x34', '\x65', '\x31', '\x65',
+				'\x32', '\x30', '\x00', '\x07', '\x6d', '\x65', '\x74', '\x61',
+				'\x64', '\x61', '\x74', '\x61', '\x00', '\x5c', '\x11', '\x51',
+				'\x56', '\xc9', '\x75', '\x50', '\x47', '\xe3', '\x18', '\xbb',
+				'\xfd', '\x00',
+			},
+			mockColl: &mockCollection{},
+		},
+		{
+			label: "UnSuccessfull Delete of entry",
+			input: map[string]interface{}{
+				"coll": "collname",
+				"key":  "keyvalue",
+				"tag":  "tagName",
+			},
+			mockColl: &mockCollection{
+				Err: pkgerrors.New("DB Error"),
+			},
+			expectedError: "DB Error",
+		},
+		{
+			label: "UnSuccessfull Delete, key not found",
+			input: map[string]interface{}{
+				"coll": "collname",
+				"key":  "keyvalue",
+				"tag":  "tagName",
+			},
+			bson: bson.Raw{
+				'\x5a', '\x00', '\x00', '\x00', '\x07', '\x5f', '\x69', '\x64',
+				'\x00', '\x5c', '\x11', '\x51', '\x56', '\x77', '\x7f', '\xf8',
+				'\x56', '\x54', '\x24', '\x8a', '\xe1', '\x02', '\x6b', '\x65',
+				'\x79', '\x00', '\x25', '\x00', '\x00', '\x00', '\x62', '\x38',
+				'\x32', '\x63', '\x34', '\x62', '\x62', '\x31', '\x2d', '\x30',
+				'\x39', '\x66', '\x66', '\x2d', '\x36', '\x30', '\x39', '\x33',
+				'\x2d', '\x34', '\x64', '\x35', '\x38', '\x2d', '\x38', '\x33',
+				'\x32', '\x37', '\x62', '\x39', '\x34', '\x65', '\x31', '\x65',
+				'\x32', '\x30', '\x00', '\x07', '\x6d', '\x65', '\x74', '\x61',
+				'\x64', '\x61', '\x74', '\x61', '\x00', '\x5c', '\x11', '\x51',
+				'\x56', '\xc9', '\x75', '\x50', '\x47', '\xe3', '\x18', '\xbb',
+				'\xfd', '\x00',
+			},
+			mockColl:      &mockCollection{},
+			expectedError: "Error finding objectID",
+		},
+		{
+			label: "Missing input fields",
+			input: map[string]interface{}{
+				"coll": "",
+				"key":  "",
+				"tag":  "",
+			},
+			expectedError: "Mandatory fields are missing",
+			mockColl:      &mockCollection{},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.label, func(t *testing.T) {
+			m, _ := NewMongoStore("name", &mongo.Database{})
+			// Override the getCollection function with our mocked version
+			getCollection = func(coll string, m *MongoStore) MongoCollection {
+				return testCase.mockColl
+			}
+
+			decodeBytes = func(sr *mongo.SingleResult) (bson.Raw, error) {
+				return testCase.bson, testCase.mockColl.Err
+			}
+			err := m.Delete(testCase.input["coll"].(string), testCase.input["key"].(string),
+				testCase.input["tag"].(string))
+			if err != nil {
+				if testCase.expectedError == "" {
+					t.Fatalf("Delete method returned an un-expected (%s)", err)
+				}
+				if !strings.Contains(string(err.Error()), testCase.expectedError) {
+					t.Fatalf("Delete method returned an error (%s)", err)
+				}
+			}
+		})
+	}
+}
+
+func TestReadAll(t *testing.T) {
+	testCases := []struct {
+		label         string
+		input         map[string]interface{}
+		mockColl      *mockCollection
+		bson          bson.Raw
+		expectedError string
+		expected      map[string][]byte
+	}{
+		{
+			label: "Successfully Read all entries",
+			input: map[string]interface{}{
+				"coll": "collname",
+				"tag":  "metadata",
+			},
+			mockColl: &mockCollection{
+				mCursor: &mockCursor{
+					// Binary form of
+					// {
+					//	"_id" : ObjectId("5c115156777ff85654248ae1"),
+					//  "key" : "b82c4bb1-09ff-6093-4d58-8327b94e1e20",
+					//  "metadata" : ObjectId("5c115156c9755047e318bbfd")
+					// }
+					bson: bson.Raw{
+						'\x5a', '\x00', '\x00', '\x00', '\x07', '\x5f', '\x69', '\x64',
+						'\x00', '\x5c', '\x11', '\x51', '\x56', '\x77', '\x7f', '\xf8',
+						'\x56', '\x54', '\x24', '\x8a', '\xe1', '\x02', '\x6b', '\x65',
+						'\x79', '\x00', '\x25', '\x00', '\x00', '\x00', '\x62', '\x38',
+						'\x32', '\x63', '\x34', '\x62', '\x62', '\x31', '\x2d', '\x30',
+						'\x39', '\x66', '\x66', '\x2d', '\x36', '\x30', '\x39', '\x33',
+						'\x2d', '\x34', '\x64', '\x35', '\x38', '\x2d', '\x38', '\x33',
+						'\x32', '\x37', '\x62', '\x39', '\x34', '\x65', '\x31', '\x65',
+						'\x32', '\x30', '\x00', '\x07', '\x6d', '\x65', '\x74', '\x61',
+						'\x64', '\x61', '\x74', '\x61', '\x00', '\x5c', '\x11', '\x51',
+						'\x56', '\xc9', '\x75', '\x50', '\x47', '\xe3', '\x18', '\xbb',
+						'\xfd', '\x00',
+					},
+					count: 1,
+				},
+			},
+			expected: map[string][]byte{
+				"b82c4bb1-09ff-6093-4d58-8327b94e1e20": []byte{
+					92, 17, 81, 86, 201, 117, 80, 71, 227, 24, 187, 253},
+			},
+		},
+		{
+			label: "UnSuccessfully Read of all entries",
+			input: map[string]interface{}{
+				"coll": "collname",
+				"tag":  "tagName",
+			},
+			mockColl: &mockCollection{
+				Err: pkgerrors.New("DB Error"),
+			},
+			expectedError: "DB Error",
+		},
+		{
+			label: "UnSuccessfull Readall, tag not found",
+			input: map[string]interface{}{
+				"coll": "collname",
+				"tag":  "tagName",
+			},
+			mockColl: &mockCollection{
+				mCursor: &mockCursor{
+					// Binary form of
+					// {
+					//	"_id" : ObjectId("5c115156777ff85654248ae1"),
+					//  "key" : "b82c4bb1-09ff-6093-4d58-8327b94e1e20",
+					//  "metadata" : ObjectId("5c115156c9755047e318bbfd")
+					// }
+					bson: bson.Raw{
+						'\x5a', '\x00', '\x00', '\x00', '\x07', '\x5f', '\x69', '\x64',
+						'\x00', '\x5c', '\x11', '\x51', '\x56', '\x77', '\x7f', '\xf8',
+						'\x56', '\x54', '\x24', '\x8a', '\xe1', '\x02', '\x6b', '\x65',
+						'\x79', '\x00', '\x25', '\x00', '\x00', '\x00', '\x62', '\x38',
+						'\x32', '\x63', '\x34', '\x62', '\x62', '\x31', '\x2d', '\x30',
+						'\x39', '\x66', '\x66', '\x2d', '\x36', '\x30', '\x39', '\x33',
+						'\x2d', '\x34', '\x64', '\x35', '\x38', '\x2d', '\x38', '\x33',
+						'\x32', '\x37', '\x62', '\x39', '\x34', '\x65', '\x31', '\x65',
+						'\x32', '\x30', '\x00', '\x07', '\x6d', '\x65', '\x74', '\x61',
+						'\x64', '\x61', '\x74', '\x61', '\x00', '\x5c', '\x11', '\x51',
+						'\x56', '\xc9', '\x75', '\x50', '\x47', '\xe3', '\x18', '\xbb',
+						'\xfd', '\x00',
+					},
+					count: 1,
+				},
+			},
+			expectedError: "Did not find any objects with tag",
+		},
+		{
+			label: "Missing input fields",
+			input: map[string]interface{}{
+				"coll": "",
+				"tag":  "",
+			},
+			expectedError: "Missing collection or tag name",
+			mockColl:      &mockCollection{},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.label, func(t *testing.T) {
+			m, _ := NewMongoStore("name", &mongo.Database{})
+			// Override the getCollection function with our mocked version
+			getCollection = func(coll string, m *MongoStore) MongoCollection {
+				return testCase.mockColl
+			}
+
+			decodeBytes = func(sr *mongo.SingleResult) (bson.Raw, error) {
+				return testCase.mockColl.mCursor.DecodeBytes()
+			}
+
+			got, err := m.ReadAll(testCase.input["coll"].(string), testCase.input["tag"].(string))
+			if err != nil {
+				if testCase.expectedError == "" {
+					t.Fatalf("Readall method returned an un-expected (%s)", err)
+				}
+				if !strings.Contains(string(err.Error()), testCase.expectedError) {
+					t.Fatalf("Readall method returned an error (%s)", err)
+				}
+			} else {
+				if reflect.DeepEqual(got, testCase.expected) == false {
+					t.Fatalf("Readall returned unexpected data: %v, expected: %v",
+						got, testCase.expected)
+				}
+			}
+		})
+	}
+}
diff --git a/src/k8splugin/db/store.go b/src/k8splugin/db/store.go
index c1a8b31f..a235597a 100644
--- a/src/k8splugin/db/store.go
+++ b/src/k8splugin/db/store.go
@@ -25,21 +25,38 @@ var DBconn Store
 
 // Store is an interface for accessing a database
 type Store interface {
+	// Returns nil if db health is good
 	HealthCheck() error
 
-	Create(string, string) error
-	Read(string) (string, error)
-	// Update(string) (string, error)
-	Delete(string) error
+	// Unmarshal implements any unmarshaling needed for the database
+	Unmarshal(inp []byte, out interface{}) error
 
-	ReadAll(string) ([]string, error)
+	// Creates a new master table with key and links data with tag and
+	// creates a pointer to the newly added data in the master table
+	Create(table, key, tag string, data interface{}) error
+
+	// Reads data for a particular key with specific tag.
+	Read(table, key, tag string) ([]byte, error)
+
+	//TODO: Update(context.Context, string, interface{}) error
+
+	// Deletes a specific tag data for key.
+	// TODO: If tag is empty, it will delete all tags under key.
+	Delete(table, key, tag string) error
+
+	// Reads all master tables and data from the specified tag in table
+	ReadAll(table, tag string) (map[string][]byte, error)
 }
 
 // CreateDBClient creates the DB client
 func CreateDBClient(dbType string) error {
 	var err error
 	switch dbType {
+	case "mongo":
+		// create a mongodb database with k8splugin as the name
+		DBconn, err = NewMongoStore("k8splugin", nil)
 	case "consul":
+		// create a consul kv store
 		DBconn, err = NewConsulStore(nil)
 	default:
 		return pkgerrors.New(dbType + "DB not supported")
diff --git a/src/k8splugin/db/store_test.go b/src/k8splugin/db/store_test.go
index 9bbe4a92..eed7065f 100644
--- a/src/k8splugin/db/store_test.go
+++ b/src/k8splugin/db/store_test.go
@@ -23,9 +23,9 @@ import (
 
 func TestCreateDBClient(t *testing.T) {
 	t.Run("Successfully create DB client", func(t *testing.T) {
-		expected := &ConsulStore{}
+		expected := &MongoStore{}
 
-		err := CreateDBClient("consul")
+		err := CreateDBClient("mongo")
 		if err != nil {
 			t.Fatalf("CreateDBClient returned an error (%s)", err)
 		}
diff --git a/src/k8splugin/db/testing.go b/src/k8splugin/db/testing.go
index 672fcbfb..4b7e6078 100644
--- a/src/k8splugin/db/testing.go
+++ b/src/k8splugin/db/testing.go
@@ -16,7 +16,8 @@ limitations under the License.
 package db
 
 import (
-	"github.com/hashicorp/consul/api"
+	"encoding/json"
+	pkgerrors "github.com/pkg/errors"
 )
 
 //Creating an embedded interface via anonymous variable
@@ -24,42 +25,45 @@ import (
 //interface even if we are not implementing all the methods in it
 type MockDB struct {
 	Store
-	Items api.KVPairs
+	Items map[string][]byte
 	Err   error
 }
 
-func (m *MockDB) Create(key string, value string) error {
+func (m *MockDB) Create(table, key, tag string, data interface{}) error {
 	return m.Err
 }
 
-func (m *MockDB) Read(key string) (string, error) {
+// MockDB uses simple JSON and not BSON
+func (m *MockDB) Unmarshal(inp []byte, out interface{}) error {
+	err := json.Unmarshal(inp, out)
+	if err != nil {
+		return pkgerrors.Wrap(err, "Unmarshaling json")
+	}
+	return nil
+}
+
+func (m *MockDB) Read(table, key, tag string) ([]byte, error) {
 	if m.Err != nil {
-		return "", m.Err
+		return nil, m.Err
 	}
 
-	for _, kvpair := range m.Items {
-		if kvpair.Key == key {
-			return string(kvpair.Value), nil
+	for k, v := range m.Items {
+		if k == key {
+			return v, nil
 		}
 	}
 
-	return "", nil
+	return nil, m.Err
 }
 
-func (m *MockDB) Delete(key string) error {
+func (m *MockDB) Delete(table, key, tag string) error {
 	return m.Err
 }
 
-func (m *MockDB) ReadAll(prefix string) ([]string, error) {
+func (m *MockDB) ReadAll(table, tag string) (map[string][]byte, error) {
 	if m.Err != nil {
-		return []string{}, m.Err
-	}
-
-	var res []string
-
-	for _, keypair := range m.Items {
-		res = append(res, keypair.Key)
+		return nil, m.Err
 	}
 
-	return res, nil
+	return m.Items, nil
 }