-
Notifications
You must be signed in to change notification settings - Fork 153
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Dave Tucker <[email protected]>
- Loading branch information
1 parent
c9769e3
commit 7ab7452
Showing
7 changed files
with
2,579 additions
and
6 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,201 @@ | ||
| package server | ||
|
|
||
| import ( | ||
| "errors" | ||
| "sync" | ||
|
|
||
| "github.com/google/uuid" | ||
| "github.com/ovn-org/libovsdb/ovsdb" | ||
| ) | ||
|
|
||
| var ( | ||
| ErrNotImplemented = errors.New("not implemented") | ||
| ) | ||
|
|
||
| // Database abstracts database operations from ovsdb | ||
| type Database interface { | ||
| CreateDatabase(name string, schema *ovsdb.DatabaseSchema) error | ||
| Exists(name string) bool | ||
| Transact(database string, operations []ovsdb.Operation) ([]ovsdb.OperationResult, ovsdb.TableUpdates) | ||
| Select(database string, table string, where []ovsdb.Condition, columns []string) ovsdb.OperationResult | ||
| Insert(dastabase string, table string, uuidName string, row map[string]interface{}) (ovsdb.OperationResult, ovsdb.TableUpdates) | ||
| Update(database, table string, where []ovsdb.Condition, row map[string]interface{}) (ovsdb.OperationResult, ovsdb.TableUpdates) | ||
| Mutate(database, table string, where []ovsdb.Condition, mutations []ovsdb.Mutation) (ovsdb.OperationResult, ovsdb.TableUpdates) | ||
| Delete(database, table string, where []ovsdb.Condition) (ovsdb.OperationResult, ovsdb.TableUpdates) | ||
| Wait(database, table string, timeout int, conditions []ovsdb.Condition, columns []string, until string, rows []map[string]interface{}) ovsdb.OperationResult | ||
| Commit(database, table string, durable bool) ovsdb.OperationResult | ||
| Abort(database, table string) ovsdb.OperationResult | ||
| Comment(database, table string, comment string) ovsdb.OperationResult | ||
| Assert(database, table, lock string) ovsdb.OperationResult | ||
| } | ||
|
|
||
| type inMemoryDatabase struct { | ||
| databases map[string]Tables | ||
| mutex sync.RWMutex | ||
| } | ||
|
|
||
| type Tables map[string]*Table | ||
|
|
||
| type Table struct { | ||
| rows map[string]interface{} | ||
| mutex sync.RWMutex | ||
| } | ||
|
|
||
| func NewInMemoryDatabase() Database { | ||
| return &inMemoryDatabase{ | ||
| databases: make(map[string]Tables), | ||
| mutex: sync.RWMutex{}, | ||
| } | ||
| } | ||
|
|
||
| func (db *inMemoryDatabase) CreateDatabase(name string, schema *ovsdb.DatabaseSchema) error { | ||
| db.mutex.Lock() | ||
| defer db.mutex.Unlock() | ||
| database := make(Tables) | ||
| for k := range schema.Tables { | ||
| database[k] = &Table{ | ||
| rows: make(map[string]interface{}), | ||
| mutex: sync.RWMutex{}, | ||
| } | ||
| } | ||
| db.databases[name] = database | ||
| return nil | ||
| } | ||
|
|
||
| func (db *inMemoryDatabase) Exists(name string) bool { | ||
| db.mutex.RLock() | ||
| defer db.mutex.RUnlock() | ||
| _, ok := db.databases[name] | ||
| return ok | ||
| } | ||
|
|
||
| func (db *inMemoryDatabase) Transact(name string, operations []ovsdb.Operation) ([]ovsdb.OperationResult, ovsdb.TableUpdates) { | ||
| results := []ovsdb.OperationResult{} | ||
| updates := make(ovsdb.TableUpdates) | ||
| for i := range operations { | ||
| switch operations[i].Op { | ||
| case ovsdb.OperationInsert: | ||
| r, tu := db.Insert(name, operations[i].Table, operations[i].UUIDName, operations[i].Row) | ||
| results = append(results, r) | ||
| if tu != nil { | ||
| updates.Merge(tu) | ||
| } | ||
| case ovsdb.OperationSelect: | ||
| r := db.Select(name, operations[i].Table, operations[i].Where, operations[i].Columns) | ||
| results = append(results, r) | ||
| case ovsdb.OperationUpdate: | ||
| r, tu := db.Update(name, operations[i].Table, operations[i].Where, operations[i].Row) | ||
| results = append(results, r) | ||
| if tu != nil { | ||
| updates.Merge(tu) | ||
| } | ||
| case ovsdb.OperationMutate: | ||
| r, tu := db.Mutate(name, operations[i].Table, operations[i].Where, operations[i].Mutations) | ||
| results = append(results, r) | ||
| if tu != nil { | ||
| updates.Merge(tu) | ||
| } | ||
| case ovsdb.OperationDelete: | ||
| r, tu := db.Delete(name, operations[i].Table, operations[i].Where) | ||
| results = append(results, r) | ||
| if tu != nil { | ||
| updates.Merge(tu) | ||
| } | ||
| case ovsdb.OperationWait: | ||
| r := db.Wait(name, operations[i].Table, operations[i].Timeout, operations[i].Where, operations[i].Columns, operations[i].Until, operations[i].Rows) | ||
| results = append(results, r) | ||
| case ovsdb.OperationCommit: | ||
| durable := operations[i].Durable | ||
| r := db.Commit(name, operations[i].Table, *durable) | ||
| results = append(results, r) | ||
| case ovsdb.OperationAbort: | ||
| r := db.Abort(name, operations[i].Table) | ||
| results = append(results, r) | ||
| case ovsdb.OperationComment: | ||
| r := db.Comment(name, operations[i].Table, *operations[i].Comment) | ||
| results = append(results, r) | ||
| case ovsdb.OperationAssert: | ||
| r := db.Assert(name, operations[i].Table, *operations[i].Lock) | ||
| results = append(results, r) | ||
| default: | ||
| return nil, updates | ||
| } | ||
| } | ||
| return results, updates | ||
| } | ||
|
|
||
| func (db *inMemoryDatabase) Insert(database string, table string, uuidName string, row map[string]interface{}) (ovsdb.OperationResult, ovsdb.TableUpdates) { | ||
| db.mutex.Lock() | ||
| defer db.mutex.Unlock() | ||
|
|
||
| var targetDb Tables | ||
| var targetTable *Table | ||
| var ok bool | ||
|
|
||
| if targetDb, ok = db.databases[database]; !ok { | ||
| return ovsdb.OperationResult{ | ||
| Count: 1, | ||
| Error: "database does not exist", | ||
| }, nil | ||
| } | ||
| if targetTable, ok = targetDb[table]; !ok { | ||
| return ovsdb.OperationResult{ | ||
| Count: 1, | ||
| Error: "table does not exist", | ||
| }, nil | ||
| } | ||
|
|
||
| targetTable.mutex.Lock() | ||
| defer targetTable.mutex.Unlock() | ||
|
|
||
| // TODO: Handle Named UUIDs and existence checks | ||
| rowUUID := uuid.NewString() | ||
| row["_uuid"] = rowUUID | ||
| targetDb[table].rows[rowUUID] = row | ||
| result := ovsdb.OperationResult{ | ||
| Count: 1, | ||
| UUID: ovsdb.UUID{GoUUID: rowUUID}, | ||
| Rows: []ovsdb.ResultRow{row}, | ||
| } | ||
| return result, ovsdb.TableUpdates{ | ||
| table: { | ||
| rowUUID: {New: &ovsdb.Row{Fields: row}, Old: nil}, | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| func (db *inMemoryDatabase) Select(database string, table string, where []ovsdb.Condition, columns []string) ovsdb.OperationResult { | ||
| return ovsdb.OperationResult{Error: ErrNotImplemented.Error()} | ||
| } | ||
|
|
||
| func (db *inMemoryDatabase) Update(database, table string, where []ovsdb.Condition, row map[string]interface{}) (ovsdb.OperationResult, ovsdb.TableUpdates) { | ||
| return ovsdb.OperationResult{Error: ErrNotImplemented.Error()}, nil | ||
| } | ||
|
|
||
| func (db *inMemoryDatabase) Mutate(database, table string, where []ovsdb.Condition, mutations []ovsdb.Mutation) (ovsdb.OperationResult, ovsdb.TableUpdates) { | ||
| return ovsdb.OperationResult{Error: ErrNotImplemented.Error()}, nil | ||
| } | ||
|
|
||
| func (db *inMemoryDatabase) Delete(database, table string, where []ovsdb.Condition) (ovsdb.OperationResult, ovsdb.TableUpdates) { | ||
| return ovsdb.OperationResult{Error: ErrNotImplemented.Error()}, nil | ||
| } | ||
|
|
||
| func (db *inMemoryDatabase) Wait(database, table string, timeout int, conditions []ovsdb.Condition, columns []string, until string, rows []map[string]interface{}) ovsdb.OperationResult { | ||
| return ovsdb.OperationResult{Error: ErrNotImplemented.Error()} | ||
| } | ||
|
|
||
| func (db *inMemoryDatabase) Commit(database, table string, durable bool) ovsdb.OperationResult { | ||
| return ovsdb.OperationResult{Error: ErrNotImplemented.Error()} | ||
| } | ||
|
|
||
| func (db *inMemoryDatabase) Abort(database, table string) ovsdb.OperationResult { | ||
| return ovsdb.OperationResult{Error: ErrNotImplemented.Error()} | ||
| } | ||
|
|
||
| func (db *inMemoryDatabase) Comment(database, table string, comment string) ovsdb.OperationResult { | ||
| return ovsdb.OperationResult{Error: ErrNotImplemented.Error()} | ||
| } | ||
|
|
||
| func (db *inMemoryDatabase) Assert(database, table, lock string) ovsdb.OperationResult { | ||
| return ovsdb.OperationResult{Error: ErrNotImplemented.Error()} | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,116 @@ | ||
| package server | ||
|
|
||
| import ( | ||
| "log" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/cenkalti/rpc2" | ||
| "github.com/ovn-org/libovsdb/ovsdb" | ||
| ) | ||
|
|
||
| // monitor represents a connection to a client where db changes | ||
| // will be reflected | ||
| type monitor struct { | ||
| id string | ||
| request map[string]*ovsdb.MonitorRequest | ||
| client *rpc2.Client | ||
| updates chan ovsdb.TableUpdates | ||
| updatesMutex sync.Mutex | ||
| stopCh chan struct{} | ||
| } | ||
|
|
||
| func newMonitor(id string, request map[string]*ovsdb.MonitorRequest, client *rpc2.Client) *monitor { | ||
| m := &monitor{ | ||
| id: id, | ||
| request: request, | ||
| client: client, | ||
| updates: make(chan ovsdb.TableUpdates, 10240), | ||
| updatesMutex: sync.Mutex{}, | ||
| stopCh: make(chan struct{}, 1), | ||
| } | ||
| go m.process() | ||
| return m | ||
| } | ||
|
|
||
| func (m *monitor) process() { | ||
| ticker := time.NewTicker(100 * time.Millisecond) | ||
| for { | ||
| select { | ||
| case <-ticker.C: | ||
| go m.sendUpdate() | ||
| case <-m.stopCh: | ||
| return | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (m *monitor) sendUpdate() { | ||
| // lock the update mutex so readers don't overlap | ||
| m.updatesMutex.Lock() | ||
| numEvents := len(m.updates) | ||
| if numEvents == 0 { | ||
| return | ||
| } | ||
| var updates []ovsdb.TableUpdates | ||
| for i := 0; i < numEvents; i++ { | ||
| updates = append(updates, <-m.updates) | ||
| } | ||
| m.updatesMutex.Unlock() | ||
| tableUpdates := make(ovsdb.TableUpdates) | ||
| for _, u := range updates { | ||
| tableUpdates.Merge(u) | ||
| } | ||
| args := []interface{}{m.id, tableUpdates} | ||
| var reply interface{} | ||
| err := m.client.Call("update", args, &reply) | ||
| if err != nil { | ||
| log.Printf("client error handling update rpc: %v", err) | ||
| } | ||
| } | ||
|
|
||
| // Enqueue will enqueue an update if it matches the tables and monitor select arguments | ||
| // we take the update by value (not reference) so we can mutate it in place before | ||
| // queuing it for dispatch | ||
| func (m *monitor) Enqueue(update ovsdb.TableUpdates) { | ||
| // remove updates for tables that we aren't watching | ||
| if len(m.request) != 0 { | ||
| m.filter(update) | ||
| } | ||
| if len(update) == 0 { | ||
| return | ||
| } | ||
| m.updates <- update | ||
| } | ||
|
|
||
| func (m *monitor) filter(update ovsdb.TableUpdates) { | ||
| // remove updates for tables that we aren't watching | ||
| if len(m.request) != 0 { | ||
| for table, u := range update { | ||
| if _, ok := m.request[table]; !ok { | ||
| delete(update, table) | ||
| } | ||
| for uuid, row := range u { | ||
| switch { | ||
| case row.Insert() && m.request[table].Select.Insert(): | ||
| case row.Modify() && m.request[table].Select.Modify(): | ||
| case row.Delete() && m.request[table].Select.Delete(): | ||
| if len(m.request[table].Columns) > 0 { | ||
| discard := true | ||
| for _, c := range m.request[table].Columns { | ||
| if _, ok := row.New.Fields[c]; ok { | ||
| discard = false | ||
| break | ||
| } | ||
| } | ||
| if discard { | ||
| delete(u, uuid) | ||
| } | ||
| } | ||
| default: | ||
| delete(u, uuid) | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.