bugfix

Author: Mikhail Podtserkovskiy

main.go

@@ -40,10 +40,13 @@ func main() {
 	}
 
 	busyNodeDuration, err := time.ParseDuration(cfg.Grid.BusyNodeDuration)
-	reservedNodeDuration, err := time.ParseDuration(cfg.Grid.BusyNodeDuration)
 	if err != nil {
 		log.Fatal("Invalid value grid.busy_node_duration in config")
 	}
+	reservedNodeDuration, err := time.ParseDuration(cfg.Grid.ReservedDuration)
+	if err != nil {
+		log.Fatal("Invalid value grid.reserved_node_duration in config")
+	}
 	storageFactory, err := invokeStorageFactory(*cfg)
 	if err != nil {
 		log.Fatalf("Can't invoke storage factory, %s", err)

storage/migrations/mysql/20170607184935-keys.sql

@@ -1,11 +1,14 @@
 
 -- +migrate Up
 
+ALTER TABLE `node` ADD COLUMN `id` BIGINT(20) NOT NULL AUTO_INCREMENT PRIMARY KEY FIRST;
+ALTER TABLE `capabilities` ADD COLUMN `id` BIGINT(20) NOT NULL AUTO_INCREMENT PRIMARY KEY FIRST;
 CREATE INDEX `status_of_updated` ON `node` (`status`,`updated`);
-CREATE INDEX `sessionId` ON `node` (`sessionId`);
-ALTER TABLE `node` ADD PRIMARY KEY (`address`);
-DROP INDEX `address` ON `node`;
 CREATE INDEX `address` ON `capabilities` (`nodeAddress`);
+CREATE INDEX `sessionId` ON `node` (`sessionId`);
+DROP INDEX `addressSetName` ON `capabilities`;
+CREATE UNIQUE INDEX `addressSetName` ON `capabilities` (`name`, `nodeAddress`, `setId`);
+# SLEEP 1;
 
 -- +migrate Down
 SIGNAL SQLSTATE '45000' SET message_text = 'Impossible down this migration';

storage/mysql/factory.go

@@ -21,6 +21,9 @@ func (f *Factory) Create(config config.Config) (pool.StorageInterface, error) {
 		return nil, err
 	}
 
+	db.SetMaxIdleConns(0) // this is the root problem! set it to 0 to remove all idle connections
+	db.SetMaxOpenConns(50) // or whatever is appropriate for your setup.
+
 	storage := NewMysqlStorage(db)
 
 	migrations := &migrate.AssetMigrationSource{

storage/mysql/storage.go

@@ -2,16 +2,17 @@ package mysql
 
 import (
 	"errors"
-	"fmt"
 	"github.com/jmoiron/sqlx"
 	"github.com/qa-dev/jsonwire-grid/pool"
 	"sort"
 	"strconv"
-	"strings"
 	"time"
+	"fmt"
+	"strings"
 )
 
 type MysqlNodeModel struct {
+	ID         string `db:"id"`
 	Type       string `db:"type"`
 	Address    string `db:"address"`
 	Status     string `db:"status"`
@@ -53,7 +54,7 @@ func (s *MysqlStorage) Add(node pool.Node) error {
 		err = errors.New("[MysqlStorage/Add] Can't insert new node: " + err.Error())
 		return err
 	}
-	_, err = s.db.Exec("DELETE FROM capabilities WHERE nodeAddress = ?", node.Address)
+	_, err = tx.Exec("DELETE FROM capabilities WHERE nodeAddress = ?", node.Address)
 	if err != nil {
 		tx.Rollback()
 		err = errors.New("[MysqlStorage/Add] Can't delete old capabilities: " + err.Error())
@@ -74,7 +75,7 @@ func (s *MysqlStorage) Add(node pool.Node) error {
 	}
 
 	for _, preparedCapability := range preparedCapabilities {
-		_, err = s.db.NamedExec(
+		_, err = tx.NamedExec(
 			"INSERT INTO capabilities (nodeAddress, setId, name, value) "+
 				"VALUES (:nodeAddress, :setId, :name, :value)",
 			preparedCapability,
@@ -144,26 +145,26 @@ func (s *MysqlStorage) ReserveAvailable(capabilities pool.Capabilities) (node po
 		err = tx.QueryRowx(
 			`
 				SELECT
+					n.id,
 					n.type,
 					n.status,
 					n.address,
 					n.sessionId,
 					n.updated,
 					n.registred
 				FROM node n
-				LEFT JOIN capabilities c ON n.address = c.nodeAddress
-				WHERE `+where+`
+				LEFT JOIN capabilities c ON n.address = c.nodeAddress AND `+where+`
 			GROUP BY c.setId
 			HAVING count(c.setId) =  `+countCapabilities+`
-			ORDER BY n.updated DESC
+			ORDER BY n.updated ASC
 			LIMIT 1
 			FOR UPDATE
 		`,
 			args...).
 			StructScan(nodeModel)
 	default:
 		err = tx.QueryRowx(
-			`SELECT n.* FROM node n WHERE `+where+` ORDER BY n.updated DESC LIMIT 1 FOR UPDATE`,
+			`SELECT n.* FROM node n WHERE `+where+` ORDER BY n.updated ASC LIMIT 1 FOR UPDATE`,
 			args...).
 			StructScan(nodeModel)
 	}