Merge branch 'master' into feature/leave-command

Author: ewoutp

main.go

@@ -204,7 +204,7 @@ func init() {
 	f.StringVar(&arangodJSPath, "server.js-dir", "/usr/share/arangodb3/js", "Path of arango JS folder")
 	f.StringVar(&rrPath, "server.rr", "", "Path of rr")
 	f.IntVar(&serverThreads, "server.threads", 0, "Adjust server.threads of each server")
-	f.StringVar(&serverStorageEngine, "server.storage-engine", "mmfiles", "Type of storage engine to use (mmfiles|rocksdb) (3.2 and up)")
+	f.StringVar(&serverStorageEngine, "server.storage-engine", "", "Type of storage engine to use (mmfiles|rocksdb) (3.2 and up)")
 	f.StringVar(&rocksDBEncryptionKeyFile, "rocksdb.encryption-keyfile", "", "Key file used for RocksDB encryption. (Enterprise Edition 3.2 and up)")
 
 	f.StringVar(&dockerEndpoint, "docker.endpoint", "unix:///var/run/docker.sock", "Endpoint used to reach the docker daemon")

service/arangod_config_builder.go

@@ -47,7 +47,7 @@ import (
 
 // createArangodConf creates an arangod.conf file in the given host directory if it does not yet exists.
 // The arangod.conf file contains all settings that are considered static for the lifetime of the server.
-func createArangodConf(log zerolog.Logger, bsCfg BootstrapConfig, myHostDir, myContainerDir, myPort string, serverType ServerType) ([]Volume, configFile, error) {
+func createArangodConf(log zerolog.Logger, bsCfg BootstrapConfig, myHostDir, myContainerDir, myPort string, serverType ServerType, features DatabaseFeatures) ([]Volume, configFile, error) {
 	hostConfFileName := filepath.Join(myHostDir, arangodConfFileName)
 	containerConfFileName := filepath.Join(myContainerDir, arangodConfFileName)
 	volumes := addVolume(nil, hostConfFileName, containerConfFileName, true)
@@ -80,8 +80,8 @@ func createArangodConf(log zerolog.Logger, bsCfg BootstrapConfig, myHostDir, myC
 		serverSection.Settings["authentication"] = "true"
 		serverSection.Settings["jwt-secret"] = bsCfg.JwtSecret
 	}
-	if bsCfg.ServerStorageEngine == "rocksdb" {
-		serverSection.Settings["storage-engine"] = "rocksdb"
+	if features.HasStorageEngineOption() {
+		serverSection.Settings["storage-engine"] = bsCfg.ServerStorageEngine
 	}
 	config := configFile{
 		serverSection,

service/bootstrap_master.go

@@ -43,6 +43,14 @@ func (s *Service) bootstrapMaster(ctx context.Context, runner Runner, config Con
 		s.log.Fatal().Msgf("Port %d is already in use", containerHTTPPort)
 	}
 
+	// Select storage engine
+	storageEngine := bsCfg.ServerStorageEngine
+	if storageEngine == "" {
+		storageEngine = s.DatabaseFeatures().DefaultStorageEngine()
+		bsCfg.ServerStorageEngine = storageEngine
+	}
+	s.log.Info().Msgf("Using storage engine '%s'", bsCfg.ServerStorageEngine)
+
 	// Create initial cluster configuration
 	hasAgent := boolFromRef(bsCfg.StartAgent, !s.mode.IsSingleMode())
 	hasDBServer := boolFromRef(bsCfg.StartDBserver, true)
@@ -55,7 +63,7 @@ func (s *Service) bootstrapMaster(ctx context.Context, runner Runner, config Con
 			hasAgent, hasDBServer, hasCoordinator, hasResilientSingle,
 			hasSyncMaster, hasSyncWorker,
 			s.IsSecure()),
-		bsCfg.AgencySize)
+		bsCfg.AgencySize, storageEngine)
 	s.learnOwnAddress = config.OwnAddress == ""
 
 	// Start HTTP listener

service/bootstrap_slave.go

@@ -103,8 +103,13 @@ func (s *Service) bootstrapSlave(peerAddress string, runner Runner, config Confi
 			s.log.Fatal().Msg("Master responsed with cluster config that does not contain my ID, please check master")
 			return
 		}
+		if result.ServerStorageEngine == "" {
+			s.log.Fatal().Msg("Master responsed with cluster config that does not contain a ServerStorageEngine, please update master first")
+			return
+		}
 		// Save cluster config
 		s.myPeers = result
+		bsCfg.ServerStorageEngine = result.ServerStorageEngine
 		break
 	}
 
@@ -149,6 +154,7 @@ func (s *Service) bootstrapSlave(peerAddress string, runner Runner, config Confi
 	}
 
 	s.log.Info().Msgf("Serving as slave with ID '%s' on %s:%d...", s.id, config.OwnAddress, s.announcePort)
+	s.log.Info().Msgf("Using storage engine '%s'", bsCfg.ServerStorageEngine)
 	s.saveSetup()
 	s.startRunning(runner, config, bsCfg)
 }

service/cluster_config.go

@@ -44,6 +44,7 @@ type ClusterConfig struct {
 	AgencySize          int        // Number of agents
 	LastModified        *time.Time `json:"LastModified,omitempty"`        // Time of last modification
 	PortOffsetIncrement int        `json:"PortOffsetIncrement,omitempty"` // Increment of port offsets for peers on same address
+	ServerStorageEngine string     `json:ServerStorageEngine,omitempty"`  // Storage engine being used
 }
 
 // PeerByID returns a peer with given id & true, or false if not found.
@@ -79,10 +80,11 @@ func (p ClusterConfig) AllAgents() []Peer {
 }
 
 // Initialize a new cluster configuration
-func (p *ClusterConfig) Initialize(initialPeer Peer, agencySize int) {
+func (p *ClusterConfig) Initialize(initialPeer Peer, agencySize int, storageEngine string) {
 	p.AllPeers = []Peer{initialPeer}
 	p.AgencySize = agencySize
 	p.PortOffsetIncrement = portOffsetIncrementNew
+	p.ServerStorageEngine = storageEngine
 	p.updateLastModified()
 }
 

service/database_features.go

@@ -0,0 +1,54 @@
+//
+// DISCLAIMER
+//
+// Copyright 2018 ArangoDB GmbH, Cologne, Germany
+//
+// 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.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+// Author Ewout Prangsma
+//
+
+package service
+
+import driver "github.com/arangodb/go-driver"
+
+// DatabaseFeatures provides information about the features provided by the
+// database in a given version.
+type DatabaseFeatures driver.Version
+
+const (
+	v32 driver.Version = "3.2.0"
+	v34 driver.Version = "3.4.0"
+)
+
+// NewDatabaseFeatures returns a new DatabaseFeatures based on
+// the given version.
+func NewDatabaseFeatures(version driver.Version) DatabaseFeatures {
+	return DatabaseFeatures(version)
+}
+
+// HasStorageEngineOption returns true when `server.storage-engine`
+// option is supported.
+func (v DatabaseFeatures) HasStorageEngineOption() bool {
+	return driver.Version(v).CompareTo(v32) >= 0
+}
+
+// DefaultStorageEngine returns the default storage engine (mmfiles|rocksdb).
+func (v DatabaseFeatures) DefaultStorageEngine() string {
+	if driver.Version(v).CompareTo(v34) >= 0 {
+		return "rocksdb"
+	}
+	return "mmfiles"
+}

service/runtime_server_manager.go

@@ -82,13 +82,16 @@ type runtimeServerManagerContext interface {
 	// IsLocalSlave returns true if this peer is running as a local slave
 	IsLocalSlave() bool
 
+	// DatabaseFeatures returns the detected database features.
+	DatabaseFeatures() DatabaseFeatures
+
 	// Stop the peer
 	Stop()
 }
 
 // startServer starts a single Arangod/Arangosync server of the given type.
 func startServer(ctx context.Context, log zerolog.Logger, runtimeContext runtimeServerManagerContext, runner Runner,
-	config Config, bsCfg BootstrapConfig, myHostAddress string, serverType ServerType, restart int) (Process, bool, error) {
+	config Config, bsCfg BootstrapConfig, myHostAddress string, serverType ServerType, features DatabaseFeatures, restart int) (Process, bool, error) {
 	myPort, err := runtimeContext.serverPort(serverType)
 	if err != nil {
 		return nil, false, maskAny(err)
@@ -145,7 +148,7 @@ func startServer(ctx context.Context, log zerolog.Logger, runtimeContext runtime
 	var containerSecretFileName string
 	if processType == ProcessTypeArangod {
 		var err error
-		confVolumes, arangodConfig, err = createArangodConf(log, bsCfg, myHostDir, myContainerDir, strconv.Itoa(myPort), serverType)
+		confVolumes, arangodConfig, err = createArangodConf(log, bsCfg, myHostDir, myContainerDir, strconv.Itoa(myPort), serverType, features)
 		if err != nil {
 			return nil, false, maskAny(err)
 		}
@@ -237,7 +240,8 @@ func (s *runtimeServerManager) runServer(ctx context.Context, log zerolog.Logger
 	for {
 		myHostAddress := myPeer.Address
 		startTime := time.Now()
-		p, portInUse, err := startServer(ctx, log, runtimeContext, runner, config, bsCfg, myHostAddress, serverType, restart)
+		features := runtimeContext.DatabaseFeatures()
+		p, portInUse, err := startServer(ctx, log, runtimeContext, runner, config, bsCfg, myHostAddress, serverType, features, restart)
 		if err != nil {
 			log.Error().Err(err).Msgf("Error while starting %s", serverType)
 			if !portInUse {

service/service.go

@@ -214,6 +214,7 @@ type Service struct {
 	runtimeServerManager  runtimeServerManager
 	runtimeClusterManager runtimeClusterManager
 	upgradeManager        UpgradeManager
+	databaseFeatures      DatabaseFeatures
 }
 
 // NewService creates a new Service instance from the given config.
@@ -257,6 +258,22 @@ const (
 	arangodJWTSecretFileName = "arangod.jwtsecret"
 )
 
+// detectDatabaseFeatures queries the database version and sets the
+// databaseFeatures field.
+func (s *Service) detectDatabaseFeatures(ctx context.Context) error {
+	v, err := s.DatabaseVersion(ctx)
+	if err != nil {
+		return maskAny(err)
+	}
+	s.databaseFeatures = NewDatabaseFeatures(v)
+	return nil
+}
+
+// DatabaseFeatures returns the detected database features.
+func (s *Service) DatabaseFeatures() DatabaseFeatures {
+	return s.databaseFeatures
+}
+
 // IsSecure returns true when the cluster is using SSL for connections, false otherwise.
 func (s *Service) IsSecure() bool {
 	if s.sslKeyFile != "" {
@@ -1177,6 +1194,17 @@ func (s *Service) Run(rootCtx context.Context, bsCfg BootstrapConfig, myPeers Cl
 	runner, s.cfg, s.allowSameDataDir = s.cfg.CreateRunner(s.log)
 	s.runner = runner
 
+	// Detect database version
+	ctx := context.Background()
+	if err := s.detectDatabaseFeatures(ctx); err != nil {
+		return errors.Wrap(err, "Failed to detect database features")
+	}
+
+	// Check storage engine
+	if err := s.validateStorageEngine(bsCfg.ServerStorageEngine, s.DatabaseFeatures()); err != nil {
+		return maskAny(err)
+	}
+
 	// Start a rotate log file time
 	if s.cfg.LogRotateInterval > 0 {
 		go s.runRotateLogFiles(rootCtx)
@@ -1186,6 +1214,13 @@ func (s *Service) Run(rootCtx context.Context, bsCfg BootstrapConfig, myPeers Cl
 	if shouldRelaunch {
 		s.myPeers = myPeers
 		s.log.Info().Msgf("Relaunching service with id '%s' on %s:%d...", s.id, s.cfg.OwnAddress, s.announcePort)
+		storageEngine, err := s.readActualStorageEngine()
+		if err != nil {
+			return maskAny(err)
+		}
+		s.myPeers.ServerStorageEngine = storageEngine
+		bsCfg.ServerStorageEngine = storageEngine
+		s.log.Info().Msgf("Using storage engine '%s'", bsCfg.ServerStorageEngine)
 		s.startHTTPServer(s.cfg)
 		wg := &sync.WaitGroup{}
 		if bsCfg.StartLocalSlaves {

service/storage_engine.go

@@ -0,0 +1,85 @@
+//
+// DISCLAIMER
+//
+// Copyright 2018 ArangoDB GmbH, Cologne, Germany
+//
+// 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.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+// Author Ewout Prangsma
+//
+
+package service
+
+import (
+	"fmt"
+	"io/ioutil"
+	"path/filepath"
+	"strings"
+)
+
+// validateStorageEngine checks if the given storage engine is a valid one.
+// Empty is still allowed.
+func (s *Service) validateStorageEngine(storageEngine string, features DatabaseFeatures) error {
+	switch storageEngine {
+	case "":
+		// Not set yet. We'll choose one later
+		return nil
+	case "mmfiles":
+		// Always OK
+		return nil
+	case "rocksdb":
+		if !features.HasStorageEngineOption() {
+			return maskAny(fmt.Errorf("RocksDB storage engine is not support for this database version"))
+		}
+		return nil
+	default:
+		return maskAny(fmt.Errorf("Unknown storage engine '%s'", storageEngine))
+	}
+}
+
+// readActualStorageEngine reads the actually used storage engine from
+// the database directory.
+func (s *Service) readActualStorageEngine() (string, error) {
+	features := s.DatabaseFeatures()
+	if !features.HasStorageEngineOption() {
+		// ENGINE file does not exist
+		return features.DefaultStorageEngine(), nil
+	}
+
+	_, _, mode := s.ClusterConfig()
+	var serverType ServerType
+	if mode.IsClusterMode() {
+		// Read engine from dbserver data directory
+		serverType = ServerTypeDBServer
+	} else if mode.IsActiveFailoverMode() {
+		// Read engine from agent data directory
+		serverType = ServerTypeAgent
+	} else {
+		// Read engine from single server data directory
+		serverType = ServerTypeSingle
+	}
+	// Get directory
+	dataDir, err := s.serverHostDir(serverType)
+	if err != nil {
+		return "", maskAny(err)
+	}
+	// Read ENGINE file
+	engine, err := ioutil.ReadFile(filepath.Join(dataDir, "data", "ENGINE"))
+	if err != nil {
+		return "", maskAny(err)
+	}
+	storageEngine := strings.ToLower(strings.TrimSpace(string(engine)))
+	return storageEngine, nil
+}