package dax import "sort" // Directive contains the instructions, sent from the Controller, which a // compute node is to follow. A Directive is typically JSON-encoded and POSTed // to a compute node's `/directive` endpoint. type Directive struct { Address Address `json:"address"` // Method describes how the compute node should handle the Directive. See // the different constants of type DirectiveMethod for how this value is // handled. Method DirectiveMethod `json:"method"` Tables []*QualifiedTable `json:"schema"` ComputeRoles []ComputeRole `json:"compute-roles"` TranslateRoles []TranslateRole `json:"translate-roles"` // The following members are used by DirectiveMethodDiff. They inlude only // those roles which have changed, as opposed to the entire role set for the // worker. ComputeRolesAdded []ComputeRole `json:"compute-roles-added"` ComputeRolesRemoved []ComputeRole `json:"compute-roles-removed"` TranslateRolesAdded []TranslateRole `json:"translate-roles-added"` TranslateRolesRemoved []TranslateRole `json:"translate-roles-removed"` Version uint64 `json:"version"` } // DirectiveVersion defines how the buildDirective step of the controller gets // the next directive version. It's important that the two methods on this // interface are not consolidated into a single step, because we use each method // as a sort of lock/unlock to ensure that only one directive (per address) is // built at a time. Since we always to the `GetCurrent()` call at the beginning // of buildDirective, if two directives are being build for the same address // concurrently, then when one of the calls `SetNext()`, the RepeatableRead // isolation level enforced on the transaction will cause the latest call to // fail since the value of version will have changed since it was first read at // the beginning of its transaction. type DirectiveVersion interface { GetCurrent(tx Transaction, addr Address) (uint64, error) SetNext(tx Transaction, addr Address, current, next uint64) error } // DirectiveMethod is used to tell the compute node how it should handle the // Directive. type DirectiveMethod string const ( // DirectiveMethodFull tells the compute node consider the Directive as the // full, complete state to which it should adhere. It should diff the // Directive with its local, cached Directive and only apply the // differences. DirectiveMethodFull DirectiveMethod = "full" // DirectiveMethodFull includes only diffs. The compute node should keep // everything about its existing state the same, and just apply the diffs in // the Directive. DirectiveMethodDiff DirectiveMethod = "diff" // DirectiveMethodReset tells the compute node to delete all of its existing // data before applying the directive. DirectiveMethodReset DirectiveMethod = "reset" // DirectiveMethodSnapshot tells the compute node that the incoming // Directive should only contain data version updates related to a snapshot // request. DirectiveMethodSnapshot DirectiveMethod = "snapshot" ) // Table returns the ID'd table from the Directive's Tables list. If it's not // found, it returns nil and a non-nil error. A nil error guarantees that the // returned table is non-nil. func (d *Directive) Table(qtid QualifiedTableID) (*QualifiedTable, error) { for _, qtbl := range d.Tables { // We can't do qtbl.QualifiedID() == qtid because the value of qtid.Name // is empty and causes the equality check to fail. Hence the .Equals() // method. if qtbl.QualifiedID().Equals(qtid) { return qtbl, nil } } return nil, NewErrTableIDDoesNotExist(qtid) } // ComputeShards returns the list of shards, for the given table, for which this // compute node is responsible. It assumes that the Directive does not contain // more than one ComputeRole for the same table; in that case, we would need to // return the union of Shards. func (d *Directive) ComputeShards(tbl TableKey) ShardNums { if d == nil || d.ComputeRoles == nil { return nil } for _, cr := range d.ComputeRoles { if cr.TableKey == tbl { return cr.Shards } } return nil } // ComputeShardsMap returns a map of table to shards. It assumes that the // Directive does not contain more than one ComputeRole for the same table; in // that case, we would need to return the union of Shards. func (d *Directive) ComputeShardsMap() map[TableKey]ShardNums { m := make(map[TableKey]ShardNums) if d == nil || d.ComputeRoles == nil { return m } for _, cr := range d.ComputeRoles { m[cr.TableKey] = cr.Shards } return m } // computeShardsMapOfMaps returns a map of TableKey to a map of ShardNum in // order to support adding and removing shards as distinct values. This map can // then be converted back to a slice of ShardNum. func (d *Directive) computeShardsMapOfMaps() map[TableKey]map[ShardNum]struct{} { m := make(map[TableKey]map[ShardNum]struct{}) if d == nil || d.ComputeRoles == nil { return m } for _, cr := range d.ComputeRoles { m[cr.TableKey] = make(map[ShardNum]struct{}) for _, shardNum := range cr.Shards { m[cr.TableKey][shardNum] = struct{}{} } } return m } // TranslatePartitionsMap returns a map of table to partitions. It assumes that // the Directive does not contain more than one TranslateRole for the same // table; in that case, we would need to return the union of Partitions. func (d *Directive) TranslatePartitionsMap() map[TableKey]PartitionNums { m := make(map[TableKey]PartitionNums) if d == nil || d.TranslateRoles == nil { return m } for _, tr := range d.TranslateRoles { // Since we added FieldVersions to the TranslateRole, it's possible for // a TranslateRole to have an empty Partitions list. In that case, we // want to exclude that from the map. if len(tr.Partitions) == 0 { continue } m[tr.TableKey] = tr.Partitions } return m } // translatePartitionsMapOfMaps returns a map of TableKey to a map of // PartitionNum in order to support adding and removing partitions as distinct // values. This map can then be converted back to a slice of PartitionNum. func (d *Directive) translatePartitionsMapOfMaps() map[TableKey]map[PartitionNum]struct{} { m := make(map[TableKey]map[PartitionNum]struct{}) if d == nil || d.TranslateRoles == nil { return m } for _, tr := range d.TranslateRoles { // Since we added FieldVersions to the TranslateRole, it's possible for // a TranslateRole to have an empty Partitions list. In that case, we // want to exclude that from the map. if len(tr.Partitions) == 0 { continue } m[tr.TableKey] = make(map[PartitionNum]struct{}) for _, partitionNum := range tr.Partitions { m[tr.TableKey][partitionNum] = struct{}{} } } return m } // translateFieldsMapOfMaps returns a map of TableKey to a map of FieldName in // order to support adding and removing fields as distinct values. This map can // then be converted back to a slice of FieldName. func (d *Directive) translateFieldsMapOfMaps() map[TableKey]map[FieldName]struct{} { m := make(map[TableKey]map[FieldName]struct{}) if d == nil || d.TranslateRoles == nil { return m } for _, tr := range d.TranslateRoles { if len(tr.Fields) == 0 { continue } m[tr.TableKey] = make(map[FieldName]struct{}) for _, fname := range tr.Fields { m[tr.TableKey][fname] = struct{}{} } } return m } // TranslateFieldsMap returns a map of table to fields. It assumes that // the Directive does not contain more than one TranslateRole for the same // table; in that case, we would need to return the union of FieldValues. func (d *Directive) TranslateFieldsMap() map[TableKey][]FieldName { m := make(map[TableKey][]FieldName) if d == nil || d.TranslateRoles == nil { return m } for _, tr := range d.TranslateRoles { if len(tr.Fields) == 0 { continue } m[tr.TableKey] = tr.Fields } return m } // IsEmpty tells whether a directive is assigning actual responsibilty // to a node or not. If the directive does not assign responsibility // for any shard or partition then it is considered empty. This is // used to determine whether we can ignore an error received from // applying this directive (an empty directive is often sent to a node // which is already down). func (d *Directive) IsEmpty() bool { for _, role := range d.ComputeRoles { if len(role.Shards) > 0 { return false } } for _, role := range d.TranslateRoles { if len(role.Partitions) > 0 { return false } } return true } // Copy returns a copy of Directive. func (d *Directive) Copy() *Directive { ret := &Directive{ Address: d.Address, Method: d.Method, Version: d.Version, } ret.Tables = append(ret.Tables, d.Tables...) ret.ComputeRoles = append(ret.ComputeRoles, d.ComputeRoles...) ret.TranslateRoles = append(ret.TranslateRoles, d.TranslateRoles...) // We intenionally do not copy the `Added` and `Removed` members because // those are not necessary to keep in the cached Directive (which just needs // to include the full Directive); they are only required when sending the // diff Directive. return ret } // ApplyDiff applies the diffs specified in diff to d. func (d *Directive) ApplyDiff(diff *Directive) *Directive { // Add any tables which are included in diff but not in d. We don't remove // tables based on a diff. for _, qtbl := range diff.Tables { if t, _ := d.Table(qtbl.QualifiedID()); t == nil { d.Tables = append(d.Tables, qtbl) } } // cmap is a map of map used to apply the directive diffs. We will convert // the final map to the ComputeRoles member in the returned Directive. cmap := d.computeShardsMapOfMaps() // Handle ComputeRolesAdded for _, crole := range diff.ComputeRolesAdded { if _, ok := cmap[crole.TableKey]; !ok { cmap[crole.TableKey] = make(map[ShardNum]struct{}) } for _, shardNum := range crole.Shards { cmap[crole.TableKey][shardNum] = struct{}{} } } // Handle ComputeRolesRemoved for _, crole := range diff.ComputeRolesRemoved { if _, ok := cmap[crole.TableKey]; !ok { continue } for _, shardNum := range crole.Shards { delete(cmap[crole.TableKey], shardNum) } } // Convert cmap back to d.ComputeRoles. croles := make([]ComputeRole, 0, len(cmap)) for tkey, smap := range cmap { shards := make([]ShardNum, 0, len(smap)) for s := range smap { shards = append(shards, s) } sort.Slice(shards, func(i, j int) bool { return shards[i] < shards[j] }) croles = append(croles, ComputeRole{ TableKey: tkey, Shards: shards, }) } // Sort croles by table. sort.Slice(croles, func(i, j int) bool { return croles[i].TableKey < croles[j].TableKey }) d.ComputeRoles = croles // tmap is a map of map used to apply the directive diffs. We will convert // the final map to the TranslateRoles member in the returned Directive. tmap := d.translatePartitionsMapOfMaps() // tmapf is a map of map, specific to translate fields, used to apply the // directive diffs. We will convert the final map to the TranslateRoles // member in the returned Directive. tmapf := d.translateFieldsMapOfMaps() // Handle TransateRolesAdded for _, trole := range diff.TranslateRolesAdded { if len(trole.Fields) > 0 { // Fields. if _, ok := tmapf[trole.TableKey]; !ok { tmapf[trole.TableKey] = make(map[FieldName]struct{}) } for _, fname := range trole.Fields { tmapf[trole.TableKey][fname] = struct{}{} } } else { // Partitions. if _, ok := tmap[trole.TableKey]; !ok { tmap[trole.TableKey] = make(map[PartitionNum]struct{}) } for _, partitionNum := range trole.Partitions { tmap[trole.TableKey][partitionNum] = struct{}{} } } } // Handle TranslateRolesRemoved for _, trole := range diff.TranslateRolesRemoved { if len(trole.Fields) > 0 { // Fields. if _, ok := tmapf[trole.TableKey]; !ok { continue } for _, fname := range trole.Fields { delete(tmapf[trole.TableKey], fname) } } else { // Partitions. if _, ok := tmap[trole.TableKey]; !ok { continue } for _, partitionNum := range trole.Partitions { delete(tmap[trole.TableKey], partitionNum) } } } // Convert tmap back to d.TranslateRoles. troles := make([]TranslateRole, 0, len(tmap)+len(tmapf)) for tkey, pmap := range tmap { partitions := make([]PartitionNum, 0, len(pmap)) for p := range pmap { partitions = append(partitions, p) } sort.Slice(partitions, func(i, j int) bool { return partitions[i] < partitions[j] }) troles = append(troles, TranslateRole{ TableKey: tkey, Partitions: partitions, }) } for tkey, fmap := range tmapf { fields := make([]FieldName, 0, len(fmap)) for f := range fmap { fields = append(fields, f) } sort.Slice(fields, func(i, j int) bool { return fields[i] < fields[j] }) troles = append(troles, TranslateRole{ TableKey: tkey, Fields: fields, }) } // Sort troles by table. sort.Slice(troles, func(i, j int) bool { return troles[i].TableKey < troles[j].TableKey }) d.TranslateRoles = troles // It doesn't really matter that we set method on the directive to be // cached, but we do it just for informational purposes. d.Method = diff.Method // Finally, be sure to use the incoming version, not the version from d. d.Version = diff.Version return d } // Directives is a sortable slice of Directive. type Directives []*Directive func (d Directives) Len() int { return len(d) } func (d Directives) Less(i, j int) bool { return d[i].Address < d[j].Address } func (d Directives) Swap(i, j int) { d[i], d[j] = d[j], d[i] }