| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you 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 org.apache.brooklyn.entity.database.mysql; |
| |
| import java.text.SimpleDateFormat; |
| import java.util.Collection; |
| import java.util.Date; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Random; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.ExecutionException; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.Semaphore; |
| import java.util.regex.Matcher; |
| import java.util.regex.Pattern; |
| |
| import org.apache.brooklyn.api.entity.Entity; |
| import org.apache.brooklyn.api.mgmt.Task; |
| import org.apache.brooklyn.api.sensor.AttributeSensor; |
| import org.apache.brooklyn.core.effector.EffectorTasks; |
| import org.apache.brooklyn.core.effector.Effectors; |
| import org.apache.brooklyn.core.effector.ssh.SshEffectorTasks; |
| import org.apache.brooklyn.core.entity.EntityPredicates; |
| import org.apache.brooklyn.core.sensor.DependentConfiguration; |
| import org.apache.brooklyn.core.sensor.Sensors; |
| import org.apache.brooklyn.entity.database.mysql.MySqlNode.ExportDumpEffector; |
| import org.apache.brooklyn.entity.software.base.SoftwareProcess; |
| import org.apache.brooklyn.location.ssh.SshMachineLocation; |
| import org.apache.brooklyn.util.core.task.DynamicTasks; |
| import org.apache.brooklyn.util.core.task.TaskTags; |
| import org.apache.brooklyn.util.core.task.ssh.SshTasks; |
| import org.apache.brooklyn.util.core.task.system.ProcessTaskFactory; |
| import org.apache.brooklyn.util.core.task.system.ProcessTaskWrapper; |
| import org.apache.brooklyn.util.exceptions.Exceptions; |
| import org.apache.brooklyn.util.os.Os; |
| import org.apache.brooklyn.util.ssh.BashCommands; |
| import org.apache.brooklyn.util.text.Identifiers; |
| import org.apache.brooklyn.util.text.StringEscapes.BashStringEscapes; |
| import org.apache.commons.io.FilenameUtils; |
| import org.apache.commons.lang3.concurrent.ConcurrentUtils; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import com.google.common.base.Joiner; |
| import com.google.common.base.Optional; |
| import com.google.common.base.Predicates; |
| import com.google.common.base.Strings; |
| import com.google.common.collect.FluentIterable; |
| import com.google.common.collect.ImmutableList; |
| import com.google.common.collect.ImmutableMap; |
| import com.google.common.collect.Iterables; |
| |
| public class InitSlaveTaskBody implements Runnable { |
| private static final String SNAPSHOT_DUMP_OPTIONS = "--skip-lock-tables --single-transaction --flush-logs --hex-blob"; |
| |
| private static final Logger log = LoggerFactory.getLogger(InitSlaveTaskBody.class); |
| |
| private final MySqlCluster cluster; |
| private final MySqlNode slave; |
| private Semaphore lock; |
| |
| public InitSlaveTaskBody(MySqlCluster cluster, MySqlNode slave, Semaphore lock) { |
| this.cluster = cluster; |
| this.slave = slave; |
| this.lock = lock; |
| } |
| |
| @Override |
| public void run() { |
| // Replication init state consists of: |
| // * initial dump (optional) |
| // * location of initial dump (could be on any of the members, optional) |
| // * bin log file name |
| // * bin log position |
| // 1. Check replication state: |
| // * Does the dump exist (and the machine where it is located) |
| // * Does the bin log exist on the master |
| // 2. If the replication state is not valid create a new one |
| // * Select a slave to dump, master if no slaves |
| // * If it's a slave do 'STOP SLAVE SQL_THREAD;' |
| // * Call mysqldump to create the snapshot |
| // * When done if a slave do 'START SLAVE SQL_THREAD;' |
| // * Get master state from the dump - grep "MASTER_LOG_POS" dump.sql. |
| // If slave get state from 'SHOW SLAVE STATUS' |
| // * Save new init info in cluster - bin log name, position, dump |
| // 3. Init Slave |
| // * transfer dump to new slave (if dump exists) |
| // * import - mysql < ~/dump.sql |
| // * change master to and start slave |
| //!!! Caveat if dumping from master and MyISAM tables are used dump may be inconsistent. |
| // * Only way around it is to lock the database while dumping (or taking a snapshot through LVM which is quicker) |
| bootstrapSlaveAsync(getValidReplicationInfo(), slave); |
| cluster.getAttribute(MySqlClusterImpl.SLAVE_ID_ADDRESS_MAPPING).put(slave.getId(), slave.getAttribute(MySqlNode.SUBNET_ADDRESS)); |
| } |
| |
| private MySqlNode getMaster() { |
| return (MySqlNode) Iterables.find(cluster.getMembers(), MySqlClusterUtils.IS_MASTER); |
| } |
| |
| private void bootstrapSlaveAsync(final Future<ReplicationSnapshot> replicationInfoFuture, final MySqlNode slave) { |
| DynamicTasks.queue("bootstrap slave replication", new Runnable() { |
| @Override |
| public void run() { |
| ReplicationSnapshot replicationSnapshot; |
| try { |
| replicationSnapshot = replicationInfoFuture.get(); |
| } catch (InterruptedException | ExecutionException e) { |
| throw Exceptions.propagate(e); |
| } |
| |
| MySqlNode master = getMaster(); |
| String masterAddress = MySqlClusterUtils.validateSqlParam(master.getAttribute(MySqlNode.SUBNET_ADDRESS)); |
| Integer masterPort = master.getAttribute(MySqlNode.MYSQL_PORT); |
| String slaveAddress = MySqlClusterUtils.validateSqlParam(slave.getAttribute(MySqlNode.SUBNET_ADDRESS)); |
| String username = MySqlClusterUtils.validateSqlParam(cluster.getConfig(MySqlCluster.SLAVE_USERNAME)); |
| String password = MySqlClusterUtils.validateSqlParam(cluster.getAttribute(MySqlCluster.SLAVE_PASSWORD)); |
| |
| if (replicationSnapshot.getEntityId() != null) { |
| Entity sourceEntity = Iterables.find(cluster.getMembers(), EntityPredicates.idEqualTo(replicationSnapshot.getEntityId())); |
| String dumpId = FilenameUtils.removeExtension(replicationSnapshot.getSnapshotPath()); |
| copyDumpAsync(sourceEntity, slave, replicationSnapshot.getSnapshotPath(), dumpId); |
| DynamicTasks.queue(Effectors.invocation(slave, MySqlNode.IMPORT_DUMP, ImmutableMap.of("path", replicationSnapshot.getSnapshotPath()))); |
| //The dump resets the password to whatever is on the source instance, reset it back. |
| //We are able to still login because privileges are not flushed, so we just set the password to the same value. |
| DynamicTasks.queue(Effectors.invocation(slave, MySqlNode.CHANGE_PASSWORD, ImmutableMap.of("password", slave.getAttribute(MySqlNode.PASSWORD)))); // |
| //Flush privileges to load new users coming from the dump |
| MySqlClusterUtils.executeSqlOnNodeAsync(slave, "FLUSH PRIVILEGES;"); |
| } |
| |
| MySqlClusterUtils.executeSqlOnNodeAsync(master, String.format( |
| "CREATE USER '%s'@'%s' IDENTIFIED BY '%s';\n" + |
| "GRANT REPLICATION SLAVE ON *.* TO '%s'@'%s';\n", |
| username, slaveAddress, password, username, slaveAddress)); |
| |
| // Executing this will unblock SERVICE_UP wait in the start effector |
| String slaveCmd = String.format( |
| "CHANGE MASTER TO " + |
| "MASTER_HOST='%s', " + |
| "MASTER_PORT=%d, " + |
| "MASTER_USER='%s', " + |
| "MASTER_PASSWORD='%s', " + |
| "MASTER_LOG_FILE='%s', " + |
| "MASTER_LOG_POS=%d;\n" + |
| "START SLAVE;\n", |
| masterAddress, masterPort, |
| username, password, |
| replicationSnapshot.getBinLogName(), |
| replicationSnapshot.getBinLogPosition()); |
| MySqlClusterUtils.executeSqlOnNodeAsync(slave, slaveCmd); |
| } |
| }); |
| } |
| |
| private void copyDumpAsync(Entity source, Entity dest, String sourceDumpPath, String dumpId) { |
| final SshMachineLocation sourceMachine = EffectorTasks.getSshMachine(source); |
| final SshMachineLocation destMachine = EffectorTasks.getSshMachine(dest); |
| |
| String sourceRunDir = source.getAttribute(MySqlNode.RUN_DIR); |
| String privateKeyFile = dumpId + ".id_rsa"; |
| final Task<String> tempKeyTask = DynamicTasks.queue(SshEffectorTasks.ssh( |
| "cd $RUN_DIR", |
| "PRIVATE_KEY=" + privateKeyFile, |
| "ssh-keygen -t rsa -N '' -f $PRIVATE_KEY -C " + dumpId + " > /dev/null", |
| "cat $PRIVATE_KEY.pub") |
| .environmentVariable("RUN_DIR", sourceRunDir) |
| .machine(sourceMachine) |
| .summary("generate private key for slave access") |
| .requiringZeroAndReturningStdout()) |
| .asTask(); |
| |
| DynamicTasks.queue("add key to authorized_keys", new Runnable() { |
| @Override |
| public void run() { |
| String publicKey = tempKeyTask.getUnchecked(); |
| DynamicTasks.queue(SshEffectorTasks.ssh(String.format( |
| "cat >> ~/.ssh/authorized_keys <<EOF\n%s\nEOF", |
| publicKey)) |
| .machine(destMachine) |
| .summary("Add key to authorized_keys") |
| .requiringExitCodeZero()); |
| } |
| }); |
| |
| final ProcessTaskWrapper<Integer> copyTask = SshEffectorTasks.ssh( |
| "cd $RUN_DIR", |
| String.format( |
| "scp -o 'BatchMode yes' -o 'StrictHostKeyChecking no' -i '%s' '%s' '%s@%s:%s/%s.sql'", |
| privateKeyFile, |
| sourceDumpPath, |
| destMachine.getUser(), |
| dest.getAttribute(MySqlNode.SUBNET_ADDRESS), |
| dest.getAttribute(MySqlNode.RUN_DIR), |
| dumpId)) |
| .environmentVariable("RUN_DIR", sourceRunDir) |
| .machine(sourceMachine) |
| .summary("copy database dump to slave") |
| .newTask(); |
| // Let next couple of tasks complete even if this one fails so that we can clean up. |
| TaskTags.markInessential(copyTask); |
| DynamicTasks.queue(copyTask); |
| |
| // Delete private key |
| DynamicTasks.queue(SshEffectorTasks.ssh( |
| "cd $RUN_DIR", |
| "rm " + privateKeyFile) |
| .environmentVariable("RUN_DIR", sourceRunDir) |
| .machine(sourceMachine) |
| .summary("remove private key")); |
| |
| DynamicTasks.queue(SshEffectorTasks.ssh(String.format( |
| "sed -i'' -e '/%s/d' ~/.ssh/authorized_keys", |
| dumpId)) |
| .machine(destMachine) |
| .summary("remove private key from authorized_keys")).asTask(); |
| |
| // The task will fail if copyTask fails, but only after the private key is deleted. |
| DynamicTasks.queue("check for successful copy", new Runnable() { |
| @Override |
| public void run() { |
| copyTask.asTask().getUnchecked(); |
| } |
| }); |
| } |
| |
| private Future<ReplicationSnapshot> getValidReplicationInfo() { |
| try { |
| try { |
| lock.acquire(); |
| } catch (InterruptedException e) { |
| throw Exceptions.propagate(e); |
| } |
| ReplicationSnapshot replicationSnapshot = getReplicationInfoMasterConfig(); |
| if (replicationSnapshot == null) { |
| replicationSnapshot = getAttributeBlocking(cluster, MySqlCluster.REPLICATION_LAST_SLAVE_SNAPSHOT); |
| } |
| if (!isReplicationInfoValid(replicationSnapshot)) { |
| final MySqlNode snapshotNode = getSnapshotNode(); |
| final String dumpName = getDumpUniqueId() + ".sql"; |
| if (MySqlClusterUtils.IS_MASTER.apply(snapshotNode)) { |
| return createMasterReplicationSnapshot(snapshotNode, dumpName); |
| } else { |
| return createSlaveReplicationSnapshot(snapshotNode, dumpName); |
| } |
| } |
| return ConcurrentUtils.constantFuture(replicationSnapshot); |
| } finally { |
| lock.release(); |
| } |
| } |
| |
| /** |
| * Rebind backwards compatibility |
| * @deprecated since 0.9.0 |
| */ |
| @Deprecated |
| private ReplicationSnapshot getReplicationInfoMasterConfig() { |
| Entity master = getMaster(); |
| AttributeSensor<String> MASTER_LOG_FILE = Sensors.newStringSensor( |
| "mysql.master.log_file", "The binary log file master is writing to"); |
| AttributeSensor<Integer> MASTER_LOG_POSITION = Sensors.newIntegerSensor( |
| "mysql.master.log_position", "The position in the log file to start replication"); |
| |
| String logFile = master.sensors().get(MASTER_LOG_FILE); |
| Integer logPos = master.sensors().get(MASTER_LOG_POSITION); |
| if(logFile != null && logPos != null) { |
| ReplicationSnapshot replicationSnapshot = new ReplicationSnapshot(null, null, logFile, logPos); |
| cluster.sensors().set(MySqlCluster.REPLICATION_LAST_SLAVE_SNAPSHOT, replicationSnapshot); |
| master.sensors().set(MASTER_LOG_FILE, null); |
| master.sensors().set(MASTER_LOG_POSITION, null); |
| return replicationSnapshot; |
| } |
| return null; |
| } |
| |
| private Future<ReplicationSnapshot> createMasterReplicationSnapshot(final MySqlNode master, final String dumpName) { |
| log.info("MySql cluster " + cluster + ": generating new replication snapshot on master node " + master + " with name " + dumpName); |
| String dumpOptions = SNAPSHOT_DUMP_OPTIONS + " --master-data=2" + getDumpDatabases(master); |
| ImmutableMap<String, String> params = ImmutableMap.of( |
| ExportDumpEffector.PATH.getName(), dumpName, |
| ExportDumpEffector.ADDITIONAL_OPTIONS.getName(), dumpOptions); |
| DynamicTasks.queue(Effectors.invocation(master, MySqlNode.EXPORT_DUMP, params)); |
| return DynamicTasks.queue("get master log info from dump", new Callable<ReplicationSnapshot>() { |
| @Override |
| public ReplicationSnapshot call() throws Exception { |
| Pattern masterInfoPattern = Pattern.compile("CHANGE MASTER TO.*MASTER_LOG_FILE\\s*=\\s*'([^']+)'.*MASTER_LOG_POS\\s*=\\s*(\\d+)"); |
| String masterInfo = DynamicTasks.queue(execSshTask(master, "grep -m1 'CHANGE MASTER TO' " + dumpName, "Extract master replication status from dump") |
| .requiringZeroAndReturningStdout()).asTask().getUnchecked(); |
| Matcher masterInfoMatcher = masterInfoPattern.matcher(masterInfo); |
| if (!masterInfoMatcher.find() || masterInfoMatcher.groupCount() != 2) { |
| throw new IllegalStateException("Master dump doesn't contain replication info: " + masterInfo); |
| } |
| String masterLogFile = masterInfoMatcher.group(1); |
| int masterLogPosition = Integer.parseInt(masterInfoMatcher.group(2)); |
| ReplicationSnapshot replicationSnapshot = new ReplicationSnapshot(master.getId(), dumpName, masterLogFile, masterLogPosition); |
| cluster.sensors().set(MySqlCluster.REPLICATION_LAST_SLAVE_SNAPSHOT, replicationSnapshot); |
| return replicationSnapshot; |
| } |
| }); |
| } |
| |
| private String getDumpDatabases(MySqlNode node) { |
| // The config will be inherited from the cluster |
| Collection<String> dumpDbs = node.config().get(MySqlCluster.SLAVE_REPLICATE_DUMP_DB); |
| if (dumpDbs != null && !dumpDbs.isEmpty()) { |
| return " --databases " + Joiner.on(' ').join(Iterables.transform(dumpDbs, BashStringEscapes.wrapBash())); |
| } else { |
| return " --all-databases"; |
| } |
| } |
| |
| private Future<ReplicationSnapshot> createSlaveReplicationSnapshot(final MySqlNode slave, final String dumpName) { |
| MySqlClusterUtils.executeSqlOnNodeAsync(slave, "STOP SLAVE SQL_THREAD;"); |
| try { |
| log.info("MySql cluster " + cluster + ": generating new replication snapshot on slave node " + slave + " with name " + dumpName); |
| String dumpOptions = SNAPSHOT_DUMP_OPTIONS + getDumpDatabases(slave); |
| ImmutableMap<String, String> params = ImmutableMap.of( |
| ExportDumpEffector.PATH.getName(), dumpName, |
| ExportDumpEffector.ADDITIONAL_OPTIONS.getName(), dumpOptions); |
| DynamicTasks.queue(Effectors.invocation(slave, MySqlNode.EXPORT_DUMP, params)); |
| return DynamicTasks.queue("get master log info from slave", new Callable<ReplicationSnapshot>() { |
| @Override |
| public ReplicationSnapshot call() throws Exception { |
| String slaveStatusRow = slave.executeScript("SHOW SLAVE STATUS \\G"); |
| Map<String, String> slaveStatus = MySqlRowParser.parseSingle(slaveStatusRow); |
| String masterLogFile = slaveStatus.get("Relay_Master_Log_File"); |
| int masterLogPosition = Integer.parseInt(slaveStatus.get("Exec_Master_Log_Pos")); |
| ReplicationSnapshot replicationSnapshot = new ReplicationSnapshot(slave.getId(), dumpName, masterLogFile, masterLogPosition); |
| cluster.sensors().set(MySqlCluster.REPLICATION_LAST_SLAVE_SNAPSHOT, replicationSnapshot); |
| return replicationSnapshot; |
| } |
| }); |
| } finally { |
| MySqlClusterUtils.executeSqlOnNodeAsync(slave, "START SLAVE SQL_THREAD;"); |
| } |
| } |
| |
| private MySqlNode getSnapshotNode() { |
| String snapshotNodeId = cluster.getConfig(MySqlCluster.REPLICATION_PREFERRED_SOURCE); |
| if (snapshotNodeId != null) { |
| Optional<Entity> preferredNode = Iterables.tryFind(cluster.getMembers(), EntityPredicates.idEqualTo(snapshotNodeId)); |
| if (preferredNode.isPresent()) { |
| return (MySqlNode) preferredNode.get(); |
| } else { |
| log.warn("MySql cluster " + this + " configured with preferred snapshot node " + snapshotNodeId + " but it's not a member. Defaulting to a random slave."); |
| } |
| } |
| return getRandomSlave(); |
| } |
| |
| private MySqlNode getRandomSlave() { |
| List<MySqlNode> slaves = getHealhtySlaves(); |
| if (slaves.size() > 0) { |
| return slaves.get(new Random().nextInt(slaves.size())); |
| } else { |
| return getMaster(); |
| } |
| } |
| |
| private ImmutableList<MySqlNode> getHealhtySlaves() { |
| return FluentIterable.from(cluster.getMembers()) |
| .filter(Predicates.not(MySqlClusterUtils.IS_MASTER)) |
| .filter(EntityPredicates.attributeEqualTo(MySqlNode.SERVICE_UP, Boolean.TRUE)) |
| .filter(MySqlNode.class) |
| .toList(); |
| } |
| |
| private boolean isReplicationInfoValid(ReplicationSnapshot replicationSnapshot) { |
| MySqlNode master = getMaster(); |
| String dataDir = Strings.nullToEmpty(master.getConfig(MySqlNode.DATA_DIR)); |
| if (!checkFileExistsOnEntity(master, Os.mergePathsUnix(dataDir, replicationSnapshot.getBinLogName()))) { |
| return false; |
| } |
| if (replicationSnapshot.getEntityId() != null) { |
| Optional<Entity> snapshotSlave = Iterables.tryFind(cluster.getChildren(), EntityPredicates.idEqualTo(replicationSnapshot.getEntityId())); |
| if (!snapshotSlave.isPresent()) { |
| log.info("MySql cluster " + cluster + " missing node " + replicationSnapshot.getEntityId() + " with last snapshot " + replicationSnapshot.getSnapshotPath() + ". Will generate new snapshot."); |
| return false; |
| } |
| if (!checkFileExistsOnEntity(snapshotSlave.get(), replicationSnapshot.getSnapshotPath())) { |
| log.info("MySql cluster " + cluster + ", node " + snapshotSlave.get() + " missing replication snapshot " + replicationSnapshot.getSnapshotPath() + ". Will generate new snapshot."); |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| private boolean checkFileExistsOnEntity(Entity entity, String path) { |
| String cmd = BashCommands.chain( |
| BashCommands.requireTest(String.format("-f \"%s\"", path), "File " + path + " doesn't exist.")); |
| String summary = "Check if file " + path + " exists"; |
| return DynamicTasks.queue(execSshTask(entity, cmd, summary).allowingNonZeroExitCode()).asTask().getUnchecked() == 0; |
| } |
| |
| private ProcessTaskFactory<Integer> execSshTask(Entity entity, String cmd, String summary) { |
| SshMachineLocation machine = EffectorTasks.getSshMachine(entity); |
| return SshTasks.newSshExecTaskFactory(machine, "cd $RUN_DIR\n" + cmd) |
| .allowingNonZeroExitCode() |
| .environmentVariable("RUN_DIR", entity.getAttribute(SoftwareProcess.RUN_DIR)) |
| .summary(summary); |
| } |
| |
| private <T> T getAttributeBlocking(Entity masterNode, AttributeSensor<T> att) { |
| return DynamicTasks.queue(DependentConfiguration.attributeWhenReady(masterNode, att)).getUnchecked(); |
| } |
| |
| private String getDumpUniqueId() { |
| return "replication-dump-" + Identifiers.makeRandomId(8) + "-" + new SimpleDateFormat("yyyy-MM-dd--HH-mm-ss").format(new Date()); |
| } |
| } |
