adding applier, instance_key, instance_key_map

go/logic/applier.go

@@ -0,0 +1,95 @@
+/*
+   Copyright 2016 GitHub Inc.
+	 See https://github.com/github/gh-osc/blob/master/LICENSE
+*/
+
+package logic
+
+import (
+	gosql "database/sql"
+	"fmt"
+	"github.com/github/gh-osc/go/base"
+	"github.com/github/gh-osc/go/mysql"
+	"github.com/github/gh-osc/go/sql"
+
+	"github.com/outbrain/golib/log"
+	"github.com/outbrain/golib/sqlutils"
+)
+
+// Applier reads data from the read-MySQL-server (typically a replica, but can be the master)
+// It is used for gaining initial status and structure, and later also follow up on progress and changelog
+type Applier struct {
+	connectionConfig *mysql.ConnectionConfig
+	db               *gosql.DB
+	migrationContext *base.MigrationContext
+}
+
+func NewApplier() *Applier {
+	return &Applier{
+		connectionConfig: base.GetMigrationContext().MasterConnectionConfig,
+		migrationContext: base.GetMigrationContext(),
+	}
+}
+
+func (this *Applier) InitDBConnections() (err error) {
+	ApplierUri := this.connectionConfig.GetDBUri(this.migrationContext.DatabaseName)
+	if this.db, _, err = sqlutils.GetDB(ApplierUri); err != nil {
+		return err
+	}
+	if err := this.validateConnection(); err != nil {
+		return err
+	}
+	return nil
+}
+
+// validateConnection issues a simple can-connect to MySQL
+func (this *Applier) validateConnection() error {
+	query := `select @@global.port`
+	var port int
+	if err := this.db.QueryRow(query).Scan(&port); err != nil {
+		return err
+	}
+	if port != this.connectionConfig.Key.Port {
+		return fmt.Errorf("Unexpected database port reported: %+v", port)
+	}
+	log.Infof("connection validated on %+v", this.connectionConfig.Key)
+	return nil
+}
+
+// CreateGhostTable creates the ghost table on the master
+func (this *Applier) CreateGhostTable() error {
+	query := fmt.Sprintf(`create /* gh-osc */ table %s.%s like %s.%s`,
+		sql.EscapeName(this.migrationContext.DatabaseName),
+		sql.EscapeName(this.migrationContext.GetGhostTableName()),
+		sql.EscapeName(this.migrationContext.DatabaseName),
+		sql.EscapeName(this.migrationContext.OriginalTableName),
+	)
+	log.Infof("Creating ghost table %s.%s",
+		sql.EscapeName(this.migrationContext.DatabaseName),
+		sql.EscapeName(this.migrationContext.GetGhostTableName()),
+	)
+	if _, err := sqlutils.ExecNoPrepare(this.db, query); err != nil {
+		return err
+	}
+	log.Infof("Table created")
+	return nil
+}
+
+// CreateGhostTable creates the ghost table on the master
+func (this *Applier) AlterGhost() error {
+	query := fmt.Sprintf(`alter /* gh-osc */ table %s.%s %s`,
+		sql.EscapeName(this.migrationContext.DatabaseName),
+		sql.EscapeName(this.migrationContext.GetGhostTableName()),
+		this.migrationContext.AlterStatement,
+	)
+	log.Infof("Altering ghost table %s.%s",
+		sql.EscapeName(this.migrationContext.DatabaseName),
+		sql.EscapeName(this.migrationContext.GetGhostTableName()),
+	)
+	log.Debugf("ALTER statement: %s", query)
+	if _, err := sqlutils.ExecNoPrepare(this.db, query); err != nil {
+		return err
+	}
+	log.Infof("Table altered")
+	return nil
+}

go/mysql/instance_key.go

@@ -0,0 +1,126 @@
+/*
+   Copyright 2015 Shlomi Noach, courtesy Booking.com
+
+   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 mysql
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+const (
+	DefaultInstancePort = 3306
+)
+
+// InstanceKey is an instance indicator, identifued by hostname and port
+type InstanceKey struct {
+	Hostname string
+	Port     int
+}
+
+const detachHint = "//"
+
+// ParseInstanceKey will parse an InstanceKey from a string representation such as 127.0.0.1:3306
+func NewRawInstanceKey(hostPort string) (*InstanceKey, error) {
+	tokens := strings.SplitN(hostPort, ":", 2)
+	if len(tokens) != 2 {
+		return nil, fmt.Errorf("Cannot parse InstanceKey from %s. Expected format is host:port", hostPort)
+	}
+	instanceKey := &InstanceKey{Hostname: tokens[0]}
+	var err error
+	if instanceKey.Port, err = strconv.Atoi(tokens[1]); err != nil {
+		return instanceKey, fmt.Errorf("Invalid port: %s", tokens[1])
+	}
+
+	return instanceKey, nil
+}
+
+// ParseRawInstanceKeyLoose will parse an InstanceKey from a string representation such as 127.0.0.1:3306.
+// The port part is optional; there will be no name resolve
+func ParseRawInstanceKeyLoose(hostPort string) (*InstanceKey, error) {
+	if !strings.Contains(hostPort, ":") {
+		return &InstanceKey{Hostname: hostPort, Port: DefaultInstancePort}, nil
+	}
+	return NewRawInstanceKey(hostPort)
+}
+
+// Equals tests equality between this key and another key
+func (this *InstanceKey) Equals(other *InstanceKey) bool {
+	if other == nil {
+		return false
+	}
+	return this.Hostname == other.Hostname && this.Port == other.Port
+}
+
+// SmallerThan returns true if this key is dictionary-smaller than another.
+// This is used for consistent sorting/ordering; there's nothing magical about it.
+func (this *InstanceKey) SmallerThan(other *InstanceKey) bool {
+	if this.Hostname < other.Hostname {
+		return true
+	}
+	if this.Hostname == other.Hostname && this.Port < other.Port {
+		return true
+	}
+	return false
+}
+
+// IsDetached returns 'true' when this hostname is logically "detached"
+func (this *InstanceKey) IsDetached() bool {
+	return strings.HasPrefix(this.Hostname, detachHint)
+}
+
+// IsValid uses simple heuristics to see whether this key represents an actual instance
+func (this *InstanceKey) IsValid() bool {
+	if this.Hostname == "_" {
+		return false
+	}
+	if this.IsDetached() {
+		return false
+	}
+	return len(this.Hostname) > 0 && this.Port > 0
+}
+
+// DetachedKey returns an instance key whose hostname is detahced: invalid, but recoverable
+func (this *InstanceKey) DetachedKey() *InstanceKey {
+	if this.IsDetached() {
+		return this
+	}
+	return &InstanceKey{Hostname: fmt.Sprintf("%s%s", detachHint, this.Hostname), Port: this.Port}
+}
+
+// ReattachedKey returns an instance key whose hostname is detahced: invalid, but recoverable
+func (this *InstanceKey) ReattachedKey() *InstanceKey {
+	if !this.IsDetached() {
+		return this
+	}
+	return &InstanceKey{Hostname: this.Hostname[len(detachHint):], Port: this.Port}
+}
+
+// StringCode returns an official string representation of this key
+func (this *InstanceKey) StringCode() string {
+	return fmt.Sprintf("%s:%d", this.Hostname, this.Port)
+}
+
+// DisplayString returns a user-friendly string representation of this key
+func (this *InstanceKey) DisplayString() string {
+	return this.StringCode()
+}
+
+// String returns a user-friendly string representation of this key
+func (this InstanceKey) String() string {
+	return this.StringCode()
+}

go/mysql/instance_key_map.go

@@ -0,0 +1,106 @@
+/*
+   Copyright 2015 Shlomi Noach, courtesy Booking.com
+
+   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 mysql
+
+import (
+	"encoding/json"
+	"strings"
+)
+
+// InstanceKeyMap is a convenience struct for listing InstanceKey-s
+type InstanceKeyMap map[InstanceKey]bool
+
+func NewInstanceKeyMap() *InstanceKeyMap {
+	return &InstanceKeyMap{}
+}
+
+// AddKey adds a single key to this map
+func (this *InstanceKeyMap) AddKey(key InstanceKey) {
+	(*this)[key] = true
+}
+
+// AddKeys adds all given keys to this map
+func (this *InstanceKeyMap) AddKeys(keys []InstanceKey) {
+	for _, key := range keys {
+		this.AddKey(key)
+	}
+}
+
+// HasKey checks if given key is within the map
+func (this *InstanceKeyMap) HasKey(key InstanceKey) bool {
+	_, ok := (*this)[key]
+	return ok
+}
+
+// GetInstanceKeys returns keys in this map in the form of an array
+func (this *InstanceKeyMap) GetInstanceKeys() []InstanceKey {
+	res := []InstanceKey{}
+	for key := range *this {
+		res = append(res, key)
+	}
+	return res
+}
+
+// MarshalJSON will marshal this map as JSON
+func (this *InstanceKeyMap) MarshalJSON() ([]byte, error) {
+	return json.Marshal(this.GetInstanceKeys())
+}
+
+// ToJSON will marshal this map as JSON
+func (this *InstanceKeyMap) ToJSON() (string, error) {
+	bytes, err := this.MarshalJSON()
+	return string(bytes), err
+}
+
+// ToJSONString will marshal this map as JSON
+func (this *InstanceKeyMap) ToJSONString() string {
+	s, _ := this.ToJSON()
+	return s
+}
+
+// ToCommaDelimitedList will export this map in comma delimited format
+func (this *InstanceKeyMap) ToCommaDelimitedList() string {
+	keyDisplays := []string{}
+	for key := range *this {
+		keyDisplays = append(keyDisplays, key.DisplayString())
+	}
+	return strings.Join(keyDisplays, ",")
+}
+
+// ReadJson unmarshalls a json into this map
+func (this *InstanceKeyMap) ReadJson(jsonString string) error {
+	var keys []InstanceKey
+	err := json.Unmarshal([]byte(jsonString), &keys)
+	if err != nil {
+		return err
+	}
+	this.AddKeys(keys)
+	return err
+}
+
+// ReadJson unmarshalls a json into this map
+func (this *InstanceKeyMap) ReadCommaDelimitedList(list string) error {
+	tokens := strings.Split(list, ",")
+	for _, token := range tokens {
+		key, err := ParseRawInstanceKeyLoose(token)
+		if err != nil {
+			return err
+		}
+		this.AddKey(*key)
+	}
+	return nil
+}