initial commit of initial GC

Signed-off-by: Alexey Roytman <roytman@il.ibm.com>

pkg/ovsdb/cache.go

@@ -95,6 +95,7 @@ func (c *cache) addDatabaseCache(dbSchema *libovsdb.DatabaseSchema, etcdClient *
 	(*c)[dbName] = dbCache
 	// we don't need etcd watcher
 	if etcdClient == nil {
+		// this is daatacache for _Server, we don't store it in etcd
 		return nil
 	}
 	ctxt := context.Background()
@@ -126,6 +127,25 @@ func (c *cache) addDatabaseCache(dbSchema *libovsdb.DatabaseSchema, etcdClient *
 			dbCache.updateCache(wresp.Events)
 		}
 	}()
+	// we have got and stored initial data, let's rig off possible GC orphances from it.
+	orphans := dbCache.initialGCCleanup()
+	if len(orphans) > 0 {
+		etcdTxn := etcdTransaction{ctx: ctx, cli: etcdClient}
+		for tableName, uuidList := range orphans {
+			for _, uuid := range uuidList {
+				key := common.NewDataKey(dbName, tableName, uuid)
+				etcdTxn.appendThen(clientv3.OpDelete(key.String()))
+			}
+		}
+		resp, err := etcdTxn.commit()
+		if err != nil {
+			log.Error(err, "orphans remove transaction failed")
+			return err
+		}
+		if resp.Succeeded {
+			log.V(5).Info("orphans remote transaction succeeded", "orphans", len(etcdTxn.then))
+		}
+	}
 	return nil
 }
 
@@ -144,6 +164,40 @@ func (dc *databaseCache) getRow(key common.Key) (cachedRow, bool) {
 	return row, ok
 }
 
+func (dc *databaseCache) initialGCCleanup() map[string][]string {
+	// [tableName][uuids]
+	orphans := map[string][]string{}
+	refCounter := refCounter{}
+	for tableName, tCache := range dc.dbCache {
+		if ! tCache.tSchema.IsRoot {
+			for uuid, cRow := range tCache.rows {
+				if cRow.counter < 1 {
+					dc.log.V(5).Info("Found an orphan", "table", tableName, "uuid", uuid, "refCounter", cRow.counter)
+					tList, ok := orphans[tableName]
+					if !ok {
+						tList = []string{}
+					}
+					tList = append(tList, uuid)
+					orphans[tableName] = tList
+					for cName, destTable := range tCache.refColumns {
+						val := cRow.row.Fields[cName]
+						if val == nil {
+							continue
+						}
+						cSchema, _ := tCache.tSchema.LookupColumn(cName)
+						checkCounters(nil, val, cSchema.Type)
+						refCounter.updateCounters(destTable, checkCounters(nil, val, cSchema.Type))
+					}
+				}
+			}
+		}
+	}
+	if len(refCounter) > 0 {
+		gc()
+	}
+	return orphans
+}
+
 func (dc *databaseCache) updateCache(events []*clientv3.Event) {
 	// we want to minimize the lock time, so we first prepare ALL the rows, and only after that update the cache
 	rows := map[common.Key]*cachedRow{}
@@ -199,36 +253,6 @@ func (dc *databaseCache) storeValues(kvs []*mvccpb.KeyValue) error {
 	return nil
 }
 
-/*
-func (dc *databaseCache) deleteCounters(newVal interface{}, oldVal interface{}, columnType string, refTable *tableCache ) {
-	if columnType == libovsdb.TypeSet {
-		valSet, ok := newVal.(libovsdb.OvsSet)
-		if !ok {
-			// TODO
-		}
-		for _, uuid := range valSet.GoSet {
-			ovsUUID := uuid.(libovsdb.UUID)
-			cRow, ok := refTable.rows[ovsUUID.GoUUID]
-			if ok {
-				cRow.counter--
-			}
-		}
-	} else if columnType == libovsdb.TypeMap {
-		valMap, ok := newVal.(libovsdb.OvsMap)
-		if !ok {
-			// TODO
-		}
-		for _, v := range valMap.GoMap {
-			ovsUUID, _ := v.(libovsdb.UUID)
-			cRow, ok := refTable.rows[ovsUUID.GoUUID]
-			if ok {
-				cRow.counter--
-			}
-		}
-	}
-}
-*/
-
 func (dc *databaseCache) updateCountersSet(newVal interface{}, oldVal interface{}, refTable *tableCache, tableKey *common.Key, newRows map[common.Key]*cachedRow) {
 	newValSet := interfaceToSet(newVal)
 	oldValSet := interfaceToSet(oldVal)

pkg/ovsdb/cache_test.go

@@ -397,3 +397,7 @@ func TestCacheCheckMap(t *testing.T) {
 	assert.Equal(t, 0, counts[val1UUID.GoUUID])
 	assert.Equal(t, 0, counts[val2UUID.GoUUID])
 }
+
+func TestInitialCounters(t *testing.T) {
+
+}

pkg/ovsdb/database.go

@@ -233,7 +233,6 @@ func (con *DatabaseEtcd) StartLeaderElection() {
 			if err != nil {
 				con.log.Error(err, "Leader Election error", "serverId", con.serverID)
 			} else {
-				con.log.V(1).Info("err is nil")
 				break
 			}
 		}

pkg/ovsdb/transact.go

@@ -646,13 +646,13 @@ func makeValue(row *map[string]interface{}) (string, error) {
 	return string(b), nil
 }
 
-func setRowUUID(row *map[string]interface{}, uuid string) {
-	(*row)[libovsdb.ColUuid] = libovsdb.UUID{GoUUID: uuid}
+func setRowUUID(row map[string]interface{}, uuid string) {
+	row[libovsdb.ColUuid] = libovsdb.UUID{GoUUID: uuid}
 }
 
-func setRowVersion(row *map[string]interface{}) {
+func setRowVersion(row map[string]interface{}) {
 	version := common.GenerateUUID()
-	(*row)[libovsdb.ColVersion] = libovsdb.UUID{GoUUID: version}
+	row[libovsdb.ColVersion] = libovsdb.UUID{GoUUID: version}
 }
 
 /*func (txn *Transaction) getUUIDIfExists(tableSchema *libovsdb.TableSchema, mapUUID namedUUIDResolver, cond1 interface{}) (string, error) {
@@ -899,7 +899,7 @@ func (txn *Transaction) doInsert(ovsOp *libovsdb.Operation, ovsResult *libovsdb.
 	key := common.NewDataKey(txn.request.DBName, *ovsOp.Table, uuid)
 	row := &map[string]interface{}{}
 
-	*row = *ovsOp.Row
+	row = ovsOp.Row
 	txn.schema.Default(*ovsOp.Table, row)
 
 	err = txn.rowPrepare(tableSchema, txn.mapUUID, ovsOp.Row)
@@ -908,8 +908,8 @@ func (txn *Transaction) doInsert(ovsOp *libovsdb.Operation, ovsResult *libovsdb.
 		err = errors.New(ErrConstraintViolation)
 		return
 	}
-	setRowUUID(row, uuid)
-	setRowVersion(row)
+	setRowUUID(*row, uuid)
+	setRowVersion(*row)
 	k := key.String()
 	val, e := makeValue(row)
 	if e != nil {
@@ -970,7 +970,7 @@ func (txn *Transaction) doModify(ovsOp *libovsdb.Operation, ovsResult *libovsdb.
 				return
 			}
 		}
-		setRowVersion(newRow)
+		setRowVersion(*newRow)
 		err = txn.etcdModifyRow([]byte(key), newRow, version)
 		if err != nil {
 			return

pkg/ovsdb/transact_test.go

@@ -2151,6 +2151,7 @@ func TestTransactCommit(t *testing.T) {
 			},
 		},
 	}
+	testEtcdCleanup(t)
 	resp := testTransact(t, req, testSchemaSimple, -1)
 	validateOperationError(t, resp, 1, 0, ErrNotSupported)
 }
@@ -2164,6 +2165,7 @@ func TestTransactAbort(t *testing.T) {
 			},
 		},
 	}
+	testEtcdCleanup(t)
 	resp := testTransact(t, req, testSchemaSimple, -1)
 	validateOperationError(t, resp, 1, 0, ErrAborted)
 }
@@ -2411,6 +2413,7 @@ func TestTransactGCUpdate(t *testing.T) {
 			},
 		},
 	}
+	testEtcdCleanup(t)
 	resp := testTransact(t, req, testSchemaGC, 3)
 	validateUpdateResult(t, resp, 1, 0, 1)
 	req = &libovsdb.Transact{
@@ -2449,6 +2452,7 @@ func TestTransactGCDelete(t *testing.T) {
 			},
 		},
 	}
+	testEtcdCleanup(t)
 	resp := testTransact(t, req, testSchemaGC, 3)
 	validateUpdateResult(t, resp, 1, 0, 1)
 	req = &libovsdb.Transact{
@@ -2469,3 +2473,37 @@ func TestTransactGCDelete(t *testing.T) {
 	validateSelectResult(t, resp, 2, 0, 0)
 	validateSelectResult(t, resp, 2, 1, 0)
 }
+
+func TestInitialGCCleanup(t *testing.T) {
+	table1 := "table1"
+	table2 := "table2"
+	dbName := "gc"
+	t1RowUUID := common.GenerateUUID()
+	t2RowUUID := common.GenerateUUID()
+	//table1RowUUID := libovsdb.UUID{GoUUID: table1UUIDNamed}
+	//table2RowUUID := libovsdb.UUID{GoUUID: table2UUIDNamed}
+
+	t1Row := map[string]interface{}{
+		"name":   "t1",
+		"refSet": libovsdb.OvsSet{GoSet: []interface{}{libovsdb.UUID{GoUUID: t2RowUUID}}},
+	}
+
+	t2Row := map[string]interface{}{
+		"name":  "t2",
+	}
+	testEtcdCleanup(t)
+	testEtcdPut(t, dbName, table1, t1RowUUID, t1Row)
+	testEtcdPut(t, dbName, table2, t2RowUUID, t2Row)
+
+	cli, err := testEtcdNewCli()
+	assert.Nil(t, err)
+	defer func() {
+		err := cli.Close()
+		assert.Nil(t, err)
+	}()
+	cache := cache{}
+	ctx := context.Background()
+	err = cache.addDatabaseCache(ctx, testSchemaGC, cli, log)
+	assert.Nil(t, err)
+
+}