| /** |
| * 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.distributedlog.admin; |
| |
| import com.google.common.base.Preconditions; |
| import com.google.common.collect.Lists; |
| import org.apache.distributedlog.DistributedLogManager; |
| import org.apache.distributedlog.LogRecordWithDLSN; |
| import org.apache.distributedlog.LogSegmentMetadata; |
| import org.apache.distributedlog.ReadUtils; |
| import org.apache.distributedlog.ZooKeeperClient; |
| import org.apache.distributedlog.ZooKeeperClientBuilder; |
| import org.apache.distributedlog.impl.BKNamespaceDriver; |
| import org.apache.distributedlog.impl.acl.ZKAccessControl; |
| import org.apache.distributedlog.exceptions.DLIllegalStateException; |
| import org.apache.distributedlog.impl.federated.FederatedZKLogMetadataStore; |
| import org.apache.distributedlog.logsegment.LogSegmentEntryStore; |
| import org.apache.distributedlog.impl.metadata.BKDLConfig; |
| import org.apache.distributedlog.metadata.DLMetadata; |
| import org.apache.distributedlog.metadata.DryrunLogSegmentMetadataStoreUpdater; |
| import org.apache.distributedlog.metadata.MetadataUpdater; |
| import org.apache.distributedlog.metadata.LogSegmentMetadataStoreUpdater; |
| import org.apache.distributedlog.namespace.DistributedLogNamespace; |
| import org.apache.distributedlog.namespace.NamespaceDriver; |
| import org.apache.distributedlog.thrift.AccessControlEntry; |
| import org.apache.distributedlog.tools.DistributedLogTool; |
| import org.apache.distributedlog.util.FutureUtils; |
| import org.apache.distributedlog.util.OrderedScheduler; |
| import org.apache.distributedlog.util.SchedulerUtils; |
| import com.twitter.util.Await; |
| import com.twitter.util.Function; |
| import com.twitter.util.Future; |
| import org.apache.bookkeeper.stats.NullStatsLogger; |
| import org.apache.bookkeeper.util.IOUtils; |
| import org.apache.commons.cli.CommandLine; |
| import org.apache.commons.cli.Options; |
| import org.apache.commons.cli.ParseException; |
| import org.apache.zookeeper.KeeperException; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.IOException; |
| import java.net.URI; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Comparator; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.SortedSet; |
| import java.util.TreeSet; |
| import java.util.concurrent.ConcurrentSkipListMap; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.LinkedBlockingQueue; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicInteger; |
| |
| /** |
| * Admin Tool for DistributedLog. |
| */ |
| public class DistributedLogAdmin extends DistributedLogTool { |
| |
| static final Logger LOG = LoggerFactory.getLogger(DistributedLogAdmin.class); |
| |
| /** |
| * Fix inprogress segment with lower ledger sequence number. |
| * |
| * @param namespace |
| * dl namespace |
| * @param metadataUpdater |
| * metadata updater. |
| * @param streamName |
| * stream name. |
| * @param verbose |
| * print verbose messages. |
| * @param interactive |
| * is confirmation needed before executing actual action. |
| * @throws IOException |
| */ |
| public static void fixInprogressSegmentWithLowerSequenceNumber(final DistributedLogNamespace namespace, |
| final MetadataUpdater metadataUpdater, |
| final String streamName, |
| final boolean verbose, |
| final boolean interactive) throws IOException { |
| DistributedLogManager dlm = namespace.openLog(streamName); |
| try { |
| List<LogSegmentMetadata> segments = dlm.getLogSegments(); |
| if (verbose) { |
| System.out.println("LogSegments for " + streamName + " : "); |
| for (LogSegmentMetadata segment : segments) { |
| System.out.println(segment.getLogSegmentSequenceNumber() + "\t: " + segment); |
| } |
| } |
| LOG.info("Get log segments for {} : {}", streamName, segments); |
| // validate log segments |
| long maxCompletedLogSegmentSequenceNumber = -1L; |
| LogSegmentMetadata inprogressSegment = null; |
| for (LogSegmentMetadata segment : segments) { |
| if (!segment.isInProgress()) { |
| maxCompletedLogSegmentSequenceNumber = Math.max(maxCompletedLogSegmentSequenceNumber, segment.getLogSegmentSequenceNumber()); |
| } else { |
| // we already found an inprogress segment |
| if (null != inprogressSegment) { |
| throw new DLIllegalStateException("Multiple inprogress segments found for stream " + streamName + " : " + segments); |
| } |
| inprogressSegment = segment; |
| } |
| } |
| if (null == inprogressSegment || inprogressSegment.getLogSegmentSequenceNumber() > maxCompletedLogSegmentSequenceNumber) { |
| // nothing to fix |
| return; |
| } |
| final long newLogSegmentSequenceNumber = maxCompletedLogSegmentSequenceNumber + 1; |
| if (interactive && !IOUtils.confirmPrompt("Confirm to fix (Y/N), Ctrl+C to break : ")) { |
| return; |
| } |
| final LogSegmentMetadata newSegment = |
| FutureUtils.result(metadataUpdater.changeSequenceNumber(inprogressSegment, newLogSegmentSequenceNumber)); |
| LOG.info("Fixed {} : {} -> {} ", |
| new Object[] { streamName, inprogressSegment, newSegment }); |
| if (verbose) { |
| System.out.println("Fixed " + streamName + " : " + inprogressSegment.getZNodeName() |
| + " -> " + newSegment.getZNodeName()); |
| System.out.println("\t old: " + inprogressSegment); |
| System.out.println("\t new: " + newSegment); |
| System.out.println(); |
| } |
| } finally { |
| dlm.close(); |
| } |
| } |
| |
| private static class LogSegmentCandidate { |
| final LogSegmentMetadata metadata; |
| final LogRecordWithDLSN lastRecord; |
| |
| LogSegmentCandidate(LogSegmentMetadata metadata, LogRecordWithDLSN lastRecord) { |
| this.metadata = metadata; |
| this.lastRecord = lastRecord; |
| } |
| |
| @Override |
| public String toString() { |
| return "LogSegmentCandidate[ metadata = " + metadata + ", last record = " + lastRecord + " ]"; |
| } |
| |
| } |
| |
| private static final Comparator<LogSegmentCandidate> LOG_SEGMENT_CANDIDATE_COMPARATOR = |
| new Comparator<LogSegmentCandidate>() { |
| @Override |
| public int compare(LogSegmentCandidate o1, LogSegmentCandidate o2) { |
| return LogSegmentMetadata.COMPARATOR.compare(o1.metadata, o2.metadata); |
| } |
| }; |
| |
| private static class StreamCandidate { |
| |
| final String streamName; |
| final SortedSet<LogSegmentCandidate> segmentCandidates = |
| new TreeSet<LogSegmentCandidate>(LOG_SEGMENT_CANDIDATE_COMPARATOR); |
| |
| StreamCandidate(String streamName) { |
| this.streamName = streamName; |
| } |
| |
| synchronized void addLogSegmentCandidate(LogSegmentCandidate segmentCandidate) { |
| segmentCandidates.add(segmentCandidate); |
| } |
| |
| @Override |
| public String toString() { |
| return "StreamCandidate[ name = " + streamName + ", segments = " + segmentCandidates + " ]"; |
| } |
| } |
| |
| public static void checkAndRepairDLNamespace(final URI uri, |
| final DistributedLogNamespace namespace, |
| final MetadataUpdater metadataUpdater, |
| final OrderedScheduler scheduler, |
| final boolean verbose, |
| final boolean interactive) throws IOException { |
| checkAndRepairDLNamespace(uri, namespace, metadataUpdater, scheduler, verbose, interactive, 1); |
| } |
| |
| public static void checkAndRepairDLNamespace(final URI uri, |
| final DistributedLogNamespace namespace, |
| final MetadataUpdater metadataUpdater, |
| final OrderedScheduler scheduler, |
| final boolean verbose, |
| final boolean interactive, |
| final int concurrency) throws IOException { |
| Preconditions.checkArgument(concurrency > 0, "Invalid concurrency " + concurrency + " found."); |
| // 0. getting streams under a given uri. |
| Iterator<String> streamsIter = namespace.getLogs(); |
| List<String> streams = Lists.newArrayList(); |
| while (streamsIter.hasNext()) { |
| streams.add(streamsIter.next()); |
| } |
| if (verbose) { |
| System.out.println("- 0. checking streams under " + uri); |
| } |
| if (streams.size() == 0) { |
| System.out.println("+ 0. nothing to check. quit."); |
| return; |
| } |
| Map<String, StreamCandidate> streamCandidates = |
| checkStreams(namespace, streams, scheduler, concurrency); |
| if (verbose) { |
| System.out.println("+ 0. " + streamCandidates.size() + " corrupted streams found."); |
| } |
| if (interactive && !IOUtils.confirmPrompt("Do you want to fix all " + streamCandidates.size() + " corrupted streams (Y/N) : ")) { |
| return; |
| } |
| if (verbose) { |
| System.out.println("- 1. repairing " + streamCandidates.size() + " corrupted streams."); |
| } |
| for (StreamCandidate candidate : streamCandidates.values()) { |
| if (!repairStream(metadataUpdater, candidate, verbose, interactive)) { |
| if (verbose) { |
| System.out.println("* 1. aborted repairing corrupted streams."); |
| } |
| return; |
| } |
| } |
| if (verbose) { |
| System.out.println("+ 1. repaired " + streamCandidates.size() + " corrupted streams."); |
| } |
| } |
| |
| private static Map<String, StreamCandidate> checkStreams( |
| final DistributedLogNamespace namespace, |
| final Collection<String> streams, |
| final OrderedScheduler scheduler, |
| final int concurrency) throws IOException { |
| final LinkedBlockingQueue<String> streamQueue = |
| new LinkedBlockingQueue<String>(); |
| streamQueue.addAll(streams); |
| final Map<String, StreamCandidate> candidateMap = |
| new ConcurrentSkipListMap<String, StreamCandidate>(); |
| final AtomicInteger numPendingStreams = new AtomicInteger(streams.size()); |
| final CountDownLatch doneLatch = new CountDownLatch(1); |
| Runnable checkRunnable = new Runnable() { |
| @Override |
| public void run() { |
| while (!streamQueue.isEmpty()) { |
| String stream; |
| try { |
| stream = streamQueue.take(); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| break; |
| } |
| StreamCandidate candidate; |
| try { |
| LOG.info("Checking stream {}.", stream); |
| candidate = checkStream(namespace, stream, scheduler); |
| LOG.info("Checked stream {} - {}.", stream, candidate); |
| } catch (IOException e) { |
| LOG.error("Error on checking stream {} : ", stream, e); |
| doneLatch.countDown(); |
| break; |
| } |
| if (null != candidate) { |
| candidateMap.put(stream, candidate); |
| } |
| if (numPendingStreams.decrementAndGet() == 0) { |
| doneLatch.countDown(); |
| } |
| } |
| } |
| }; |
| Thread[] threads = new Thread[concurrency]; |
| for (int i = 0; i < concurrency; i++) { |
| threads[i] = new Thread(checkRunnable, "check-thread-" + i); |
| threads[i].start(); |
| } |
| try { |
| doneLatch.await(); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| if (numPendingStreams.get() != 0) { |
| throw new IOException(numPendingStreams.get() + " streams left w/o checked"); |
| } |
| for (int i = 0; i < concurrency; i++) { |
| threads[i].interrupt(); |
| try { |
| threads[i].join(); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| } |
| return candidateMap; |
| } |
| |
| private static StreamCandidate checkStream( |
| final DistributedLogNamespace namespace, |
| final String streamName, |
| final OrderedScheduler scheduler) throws IOException { |
| DistributedLogManager dlm = namespace.openLog(streamName); |
| try { |
| List<LogSegmentMetadata> segments = dlm.getLogSegments(); |
| if (segments.isEmpty()) { |
| return null; |
| } |
| List<Future<LogSegmentCandidate>> futures = |
| new ArrayList<Future<LogSegmentCandidate>>(segments.size()); |
| for (LogSegmentMetadata segment : segments) { |
| futures.add(checkLogSegment(namespace, streamName, segment, scheduler)); |
| } |
| List<LogSegmentCandidate> segmentCandidates; |
| try { |
| segmentCandidates = Await.result(Future.collect(futures)); |
| } catch (Exception e) { |
| throw new IOException("Failed on checking stream " + streamName, e); |
| } |
| StreamCandidate streamCandidate = new StreamCandidate(streamName); |
| for (LogSegmentCandidate segmentCandidate: segmentCandidates) { |
| if (null != segmentCandidate) { |
| streamCandidate.addLogSegmentCandidate(segmentCandidate); |
| } |
| } |
| if (streamCandidate.segmentCandidates.isEmpty()) { |
| return null; |
| } |
| return streamCandidate; |
| } finally { |
| dlm.close(); |
| } |
| } |
| |
| private static Future<LogSegmentCandidate> checkLogSegment( |
| final DistributedLogNamespace namespace, |
| final String streamName, |
| final LogSegmentMetadata metadata, |
| final OrderedScheduler scheduler) { |
| if (metadata.isInProgress()) { |
| return Future.value(null); |
| } |
| |
| final LogSegmentEntryStore entryStore = namespace.getNamespaceDriver() |
| .getLogSegmentEntryStore(NamespaceDriver.Role.READER); |
| return ReadUtils.asyncReadLastRecord( |
| streamName, |
| metadata, |
| true, |
| false, |
| true, |
| 4, |
| 16, |
| new AtomicInteger(0), |
| scheduler, |
| entryStore |
| ).map(new Function<LogRecordWithDLSN, LogSegmentCandidate>() { |
| @Override |
| public LogSegmentCandidate apply(LogRecordWithDLSN record) { |
| if (null != record && |
| (record.getDlsn().compareTo(metadata.getLastDLSN()) > 0 || |
| record.getTransactionId() > metadata.getLastTxId() || |
| !metadata.isRecordPositionWithinSegmentScope(record))) { |
| return new LogSegmentCandidate(metadata, record); |
| } else { |
| return null; |
| } |
| } |
| }); |
| } |
| |
| private static boolean repairStream(MetadataUpdater metadataUpdater, |
| StreamCandidate streamCandidate, |
| boolean verbose, |
| boolean interactive) throws IOException { |
| if (verbose) { |
| System.out.println("Stream " + streamCandidate.streamName + " : "); |
| for (LogSegmentCandidate segmentCandidate : streamCandidate.segmentCandidates) { |
| System.out.println(" " + segmentCandidate.metadata.getLogSegmentSequenceNumber() |
| + " : metadata = " + segmentCandidate.metadata + ", last dlsn = " |
| + segmentCandidate.lastRecord.getDlsn()); |
| } |
| System.out.println("-------------------------------------------"); |
| } |
| if (interactive && !IOUtils.confirmPrompt("Do you want to fix the stream " + streamCandidate.streamName + " (Y/N) : ")) { |
| return false; |
| } |
| for (LogSegmentCandidate segmentCandidate : streamCandidate.segmentCandidates) { |
| LogSegmentMetadata newMetadata = FutureUtils.result( |
| metadataUpdater.updateLastRecord(segmentCandidate.metadata, segmentCandidate.lastRecord)); |
| if (verbose) { |
| System.out.println(" Fixed segment " + segmentCandidate.metadata.getLogSegmentSequenceNumber() + " : "); |
| System.out.println(" old metadata : " + segmentCandidate.metadata); |
| System.out.println(" new metadata : " + newMetadata); |
| } |
| } |
| if (verbose) { |
| System.out.println("-------------------------------------------"); |
| } |
| return true; |
| } |
| |
| // |
| // Commands |
| // |
| |
| /** |
| * Unbind the bookkeeper environment for a given distributedlog uri. |
| * |
| * TODO: move unbind operation to namespace driver |
| */ |
| class UnbindCommand extends OptsCommand { |
| |
| Options options = new Options(); |
| |
| UnbindCommand() { |
| super("unbind", "unbind the bookkeeper environment bound for a given distributedlog instance."); |
| options.addOption("f", "force", false, "Force unbinding without prompt."); |
| } |
| |
| @Override |
| protected Options getOptions() { |
| return options; |
| } |
| |
| @Override |
| protected String getUsage() { |
| return "unbind [options] <distributedlog uri>"; |
| } |
| |
| @Override |
| protected int runCmd(CommandLine cmdline) throws Exception { |
| String[] args = cmdline.getArgs(); |
| if (args.length <= 0) { |
| System.err.println("No distributedlog uri specified."); |
| printUsage(); |
| return -1; |
| } |
| boolean force = cmdline.hasOption("f"); |
| URI uri = URI.create(args[0]); |
| // resolving the uri to see if there is another bindings in this uri. |
| ZooKeeperClient zkc = ZooKeeperClientBuilder.newBuilder().uri(uri) |
| .sessionTimeoutMs(10000).build(); |
| BKDLConfig bkdlConfig; |
| try { |
| bkdlConfig = BKDLConfig.resolveDLConfig(zkc, uri); |
| } catch (IOException ie) { |
| bkdlConfig = null; |
| } |
| if (null == bkdlConfig) { |
| System.out.println("No bookkeeper is bound to " + uri); |
| return 0; |
| } else { |
| System.out.println("There is bookkeeper bound to " + uri + " : "); |
| System.out.println(""); |
| System.out.println(bkdlConfig.toString()); |
| System.out.println(""); |
| if (!force && !IOUtils.confirmPrompt("Do you want to unbind " + uri + " :\n")) { |
| return 0; |
| } |
| } |
| DLMetadata.unbind(uri); |
| System.out.println("Unbound on " + uri + "."); |
| return 0; |
| } |
| } |
| |
| /** |
| * Bind Command to bind bookkeeper environment for a given distributed uri. |
| * |
| * TODO: move bind to namespace driver |
| */ |
| class BindCommand extends OptsCommand { |
| |
| Options options = new Options(); |
| |
| BindCommand() { |
| super("bind", "bind the bookkeeper environment settings for a given distributedlog instance."); |
| options.addOption("l", "bkLedgers", true, "ZooKeeper ledgers path for bookkeeper instance."); |
| options.addOption("s", "bkZkServers", true, "ZooKeeper servers used for bookkeeper for writers."); |
| options.addOption("bkzr", "bkZkServersForReader", true, "ZooKeeper servers used for bookkeeper for readers."); |
| options.addOption("dlzw", "dlZkServersForWriter", true, "ZooKeeper servers used for distributedlog for writers."); |
| options.addOption("dlzr", "dlZkServersForReader", true, "ZooKeeper servers used for distributedlog for readers."); |
| options.addOption("i", "sanityCheckTxnID", true, "Flag to sanity check highest txn id."); |
| options.addOption("r", "encodeRegionID", true, "Flag to encode region id."); |
| options.addOption("seqno", "firstLogSegmentSeqNo", true, "The first log segment sequence number to use after upgrade"); |
| options.addOption("fns", "federatedNamespace", false, "Flag to turn a namespace to federated namespace"); |
| options.addOption("f", "force", false, "Force binding without prompt."); |
| options.addOption("c", "creation", false, "Whether is it a creation binding."); |
| options.addOption("q", "query", false, "Query the bookkeeper bindings"); |
| } |
| |
| @Override |
| protected Options getOptions() { |
| return options; |
| } |
| |
| @Override |
| protected String getUsage() { |
| return "bind [options] <distributedlog uri>"; |
| } |
| |
| @Override |
| protected int runCmd(CommandLine cmdline) throws Exception { |
| boolean isQuery = cmdline.hasOption("q"); |
| if (!isQuery && (!cmdline.hasOption("l") || !cmdline.hasOption("s"))) { |
| System.err.println("Error: Neither zkServers nor ledgersPath specified for bookkeeper environment."); |
| printUsage(); |
| return -1; |
| } |
| String[] args = cmdline.getArgs(); |
| if (args.length <= 0) { |
| System.err.println("No distributedlog uri specified."); |
| printUsage(); |
| return -1; |
| } |
| boolean force = cmdline.hasOption("f"); |
| boolean creation = cmdline.hasOption("c"); |
| String bkLedgersPath = cmdline.getOptionValue("l"); |
| String bkZkServersForWriter = cmdline.getOptionValue("s"); |
| boolean sanityCheckTxnID = |
| !cmdline.hasOption("i") || Boolean.parseBoolean(cmdline.getOptionValue("i")); |
| boolean encodeRegionID = |
| cmdline.hasOption("r") && Boolean.parseBoolean(cmdline.getOptionValue("r")); |
| |
| String bkZkServersForReader; |
| if (cmdline.hasOption("bkzr")) { |
| bkZkServersForReader = cmdline.getOptionValue("bkzr"); |
| } else { |
| bkZkServersForReader = bkZkServersForWriter; |
| } |
| |
| URI uri = URI.create(args[0]); |
| |
| String dlZkServersForWriter; |
| String dlZkServersForReader; |
| if (cmdline.hasOption("dlzw")) { |
| dlZkServersForWriter = cmdline.getOptionValue("dlzw"); |
| } else { |
| dlZkServersForWriter = BKNamespaceDriver.getZKServersFromDLUri(uri); |
| } |
| if (cmdline.hasOption("dlzr")) { |
| dlZkServersForReader = cmdline.getOptionValue("dlzr"); |
| } else { |
| dlZkServersForReader = dlZkServersForWriter; |
| } |
| |
| // resolving the uri to see if there is another bindings in this uri. |
| ZooKeeperClient zkc = ZooKeeperClientBuilder.newBuilder().uri(uri).zkAclId(null) |
| .sessionTimeoutMs(10000).build(); |
| try { |
| BKDLConfig newBKDLConfig = |
| new BKDLConfig(dlZkServersForWriter, dlZkServersForReader, |
| bkZkServersForWriter, bkZkServersForReader, bkLedgersPath) |
| .setSanityCheckTxnID(sanityCheckTxnID) |
| .setEncodeRegionID(encodeRegionID); |
| |
| if (cmdline.hasOption("seqno")) { |
| newBKDLConfig = newBKDLConfig.setFirstLogSegmentSeqNo(Long.parseLong(cmdline.getOptionValue("seqno"))); |
| } |
| |
| if (cmdline.hasOption("fns")) { |
| newBKDLConfig = newBKDLConfig.setFederatedNamespace(true); |
| } |
| |
| BKDLConfig bkdlConfig; |
| try { |
| bkdlConfig = BKDLConfig.resolveDLConfig(zkc, uri); |
| } catch (IOException ie) { |
| bkdlConfig = null; |
| } |
| if (null == bkdlConfig) { |
| System.out.println("No bookkeeper is bound to " + uri); |
| } else { |
| System.out.println("There is bookkeeper bound to " + uri + " : "); |
| System.out.println(""); |
| System.out.println(bkdlConfig.toString()); |
| System.out.println(""); |
| if (!isQuery) { |
| if (newBKDLConfig.equals(bkdlConfig)) { |
| System.out.println("No bookkeeper binding needs to be updated. Quit."); |
| return 0; |
| } else if(!newBKDLConfig.isFederatedNamespace() && bkdlConfig.isFederatedNamespace()) { |
| System.out.println("You can't turn a federated namespace back to non-federated."); |
| return 0; |
| } else { |
| if (!force && !IOUtils.confirmPrompt("Do you want to bind " + uri |
| + " with new bookkeeper instance :\n" + newBKDLConfig)) { |
| return 0; |
| } |
| } |
| } |
| } |
| if (isQuery) { |
| System.out.println("Done."); |
| return 0; |
| } |
| DLMetadata dlMetadata = DLMetadata.create(newBKDLConfig); |
| if (creation) { |
| try { |
| dlMetadata.create(uri); |
| System.out.println("Created binding on " + uri + "."); |
| } catch (IOException ie) { |
| System.err.println("Failed to create binding on " + uri + " : " + ie.getMessage()); |
| } |
| } else { |
| try { |
| dlMetadata.update(uri); |
| System.out.println("Updated binding on " + uri + " : "); |
| System.out.println(""); |
| System.out.println(newBKDLConfig.toString()); |
| System.out.println(""); |
| } catch (IOException ie) { |
| System.err.println("Failed to update binding on " + uri + " : " + ie.getMessage()); |
| } |
| } |
| if (newBKDLConfig.isFederatedNamespace()) { |
| try { |
| FederatedZKLogMetadataStore.createFederatedNamespace(uri, zkc); |
| } catch (KeeperException.NodeExistsException nee) { |
| // ignore node exists exception |
| } |
| } |
| return 0; |
| } finally { |
| zkc.close(); |
| } |
| } |
| } |
| |
| static class RepairSeqNoCommand extends PerDLCommand { |
| |
| boolean dryrun = false; |
| boolean verbose = false; |
| final List<String> streams = new ArrayList<String>(); |
| |
| RepairSeqNoCommand() { |
| super("repairseqno", "Repair a stream whose inprogress log segment has lower sequence number."); |
| options.addOption("d", "dryrun", false, "Dry run without repairing"); |
| options.addOption("l", "list", true, "List of streams to repair, separated by comma"); |
| options.addOption("v", "verbose", false, "Print verbose messages"); |
| } |
| |
| @Override |
| protected void parseCommandLine(CommandLine cmdline) throws ParseException { |
| super.parseCommandLine(cmdline); |
| dryrun = cmdline.hasOption("d"); |
| verbose = cmdline.hasOption("v"); |
| force = !dryrun && cmdline.hasOption("f"); |
| if (!cmdline.hasOption("l")) { |
| throw new ParseException("No streams provided to repair"); |
| } |
| String streamsList = cmdline.getOptionValue("l"); |
| Collections.addAll(streams, streamsList.split(",")); |
| } |
| |
| @Override |
| protected int runCmd() throws Exception { |
| MetadataUpdater metadataUpdater = dryrun ? |
| new DryrunLogSegmentMetadataStoreUpdater(getConf(), |
| getLogSegmentMetadataStore()) : |
| LogSegmentMetadataStoreUpdater.createMetadataUpdater(getConf(), |
| getLogSegmentMetadataStore()); |
| System.out.println("List of streams : "); |
| System.out.println(streams); |
| if (!IOUtils.confirmPrompt("Do you want to repair all these streams (Y/N):")) { |
| return -1; |
| } |
| for (String stream : streams) { |
| fixInprogressSegmentWithLowerSequenceNumber(getNamespace(), metadataUpdater, stream, verbose, !getForce()); |
| } |
| return 0; |
| } |
| |
| @Override |
| protected String getUsage() { |
| return "repairseqno [options]"; |
| } |
| } |
| |
| static class DLCKCommand extends PerDLCommand { |
| |
| boolean dryrun = false; |
| boolean verbose = false; |
| int concurrency = 1; |
| |
| DLCKCommand() { |
| super("dlck", "Check and repair a distributedlog namespace"); |
| options.addOption("d", "dryrun", false, "Dry run without repairing"); |
| options.addOption("v", "verbose", false, "Print verbose messages"); |
| options.addOption("cy", "concurrency", true, "Concurrency on checking streams"); |
| } |
| |
| @Override |
| protected void parseCommandLine(CommandLine cmdline) throws ParseException { |
| super.parseCommandLine(cmdline); |
| dryrun = cmdline.hasOption("d"); |
| verbose = cmdline.hasOption("v"); |
| if (cmdline.hasOption("cy")) { |
| try { |
| concurrency = Integer.parseInt(cmdline.getOptionValue("cy")); |
| } catch (NumberFormatException nfe) { |
| throw new ParseException("Invalid concurrency value : " + cmdline.getOptionValue("cy")); |
| } |
| } |
| } |
| |
| @Override |
| protected int runCmd() throws Exception { |
| MetadataUpdater metadataUpdater = dryrun ? |
| new DryrunLogSegmentMetadataStoreUpdater(getConf(), |
| getLogSegmentMetadataStore()) : |
| LogSegmentMetadataStoreUpdater.createMetadataUpdater(getConf(), |
| getLogSegmentMetadataStore()); |
| OrderedScheduler scheduler = OrderedScheduler.newBuilder() |
| .name("dlck-scheduler") |
| .corePoolSize(Runtime.getRuntime().availableProcessors()) |
| .build(); |
| ExecutorService executorService = Executors.newCachedThreadPool(); |
| try { |
| checkAndRepairDLNamespace(getUri(), getNamespace(), metadataUpdater, scheduler, |
| verbose, !getForce(), concurrency); |
| } finally { |
| SchedulerUtils.shutdownScheduler(executorService, 5, TimeUnit.MINUTES); |
| } |
| return 0; |
| } |
| |
| @Override |
| protected String getUsage() { |
| return "dlck [options]"; |
| } |
| } |
| |
| static class DeleteStreamACLCommand extends PerDLCommand { |
| |
| String stream = null; |
| |
| DeleteStreamACLCommand() { |
| super("delete_stream_acl", "Delete ACL for a given stream"); |
| options.addOption("s", "stream", true, "Stream to set ACL"); |
| } |
| |
| @Override |
| protected void parseCommandLine(CommandLine cmdline) throws ParseException { |
| super.parseCommandLine(cmdline); |
| if (!cmdline.hasOption("s")) { |
| throw new ParseException("No stream to set ACL"); |
| } |
| stream = cmdline.getOptionValue("s"); |
| } |
| |
| @Override |
| protected int runCmd() throws Exception { |
| BKDLConfig bkdlConfig = BKDLConfig.resolveDLConfig(getZooKeeperClient(), getUri()); |
| if (null == bkdlConfig.getACLRootPath()) { |
| // acl isn't enabled for this namespace. |
| System.err.println("ACL isn't enabled for namespace " + getUri()); |
| return -1; |
| } |
| String zkPath = getUri() + "/" + bkdlConfig.getACLRootPath() + "/" + stream; |
| ZKAccessControl.delete(getZooKeeperClient(), zkPath); |
| return 0; |
| } |
| |
| @Override |
| protected String getUsage() { |
| return null; |
| } |
| } |
| |
| static class SetStreamACLCommand extends SetACLCommand { |
| |
| String stream = null; |
| |
| SetStreamACLCommand() { |
| super("set_stream_acl", "Set Default ACL for a given stream"); |
| options.addOption("s", "stream", true, "Stream to set ACL"); |
| } |
| |
| @Override |
| protected void parseCommandLine(CommandLine cmdline) throws ParseException { |
| super.parseCommandLine(cmdline); |
| if (!cmdline.hasOption("s")) { |
| throw new ParseException("No stream to set ACL"); |
| } |
| stream = cmdline.getOptionValue("s"); |
| } |
| |
| @Override |
| protected String getZKPath(String zkRootPath) { |
| return zkRootPath + "/" + stream; |
| } |
| |
| @Override |
| protected String getUsage() { |
| return "set_stream_acl [options]"; |
| } |
| } |
| |
| static class SetDefaultACLCommand extends SetACLCommand { |
| |
| SetDefaultACLCommand() { |
| super("set_default_acl", "Set Default ACL for a namespace"); |
| } |
| |
| @Override |
| protected String getZKPath(String zkRootPath) { |
| return zkRootPath; |
| } |
| |
| @Override |
| protected String getUsage() { |
| return "set_default_acl [options]"; |
| } |
| } |
| |
| static abstract class SetACLCommand extends PerDLCommand { |
| |
| boolean denyWrite = false; |
| boolean denyTruncate = false; |
| boolean denyDelete = false; |
| boolean denyAcquire = false; |
| boolean denyRelease = false; |
| |
| protected SetACLCommand(String name, String description) { |
| super(name, description); |
| options.addOption("dw", "deny-write", false, "Deny write/bulkWrite requests"); |
| options.addOption("dt", "deny-truncate", false, "Deny truncate requests"); |
| options.addOption("dd", "deny-delete", false, "Deny delete requests"); |
| options.addOption("da", "deny-acquire", false, "Deny acquire requests"); |
| options.addOption("dr", "deny-release", false, "Deny release requests"); |
| } |
| |
| @Override |
| protected void parseCommandLine(CommandLine cmdline) throws ParseException { |
| super.parseCommandLine(cmdline); |
| denyWrite = cmdline.hasOption("dw"); |
| denyTruncate = cmdline.hasOption("dt"); |
| denyDelete = cmdline.hasOption("dd"); |
| denyAcquire = cmdline.hasOption("da"); |
| denyRelease = cmdline.hasOption("dr"); |
| } |
| |
| protected abstract String getZKPath(String zkRootPath); |
| |
| protected ZKAccessControl getZKAccessControl(ZooKeeperClient zkc, String zkPath) throws Exception { |
| ZKAccessControl accessControl; |
| try { |
| accessControl = Await.result(ZKAccessControl.read(zkc, zkPath, null)); |
| } catch (KeeperException.NoNodeException nne) { |
| accessControl = new ZKAccessControl(new AccessControlEntry(), zkPath); |
| } |
| return accessControl; |
| } |
| |
| protected void setZKAccessControl(ZooKeeperClient zkc, ZKAccessControl accessControl) throws Exception { |
| String zkPath = accessControl.getZKPath(); |
| if (null == zkc.get().exists(zkPath, false)) { |
| accessControl.create(zkc); |
| } else { |
| accessControl.update(zkc); |
| } |
| } |
| |
| @Override |
| protected int runCmd() throws Exception { |
| BKDLConfig bkdlConfig = BKDLConfig.resolveDLConfig(getZooKeeperClient(), getUri()); |
| if (null == bkdlConfig.getACLRootPath()) { |
| // acl isn't enabled for this namespace. |
| System.err.println("ACL isn't enabled for namespace " + getUri()); |
| return -1; |
| } |
| String zkPath = getZKPath(getUri().getPath() + "/" + bkdlConfig.getACLRootPath()); |
| ZKAccessControl accessControl = getZKAccessControl(getZooKeeperClient(), zkPath); |
| AccessControlEntry acl = accessControl.getAccessControlEntry(); |
| acl.setDenyWrite(denyWrite); |
| acl.setDenyTruncate(denyTruncate); |
| acl.setDenyDelete(denyDelete); |
| acl.setDenyAcquire(denyAcquire); |
| acl.setDenyRelease(denyRelease); |
| setZKAccessControl(getZooKeeperClient(), accessControl); |
| return 0; |
| } |
| |
| } |
| |
| public DistributedLogAdmin() { |
| super(); |
| commands.clear(); |
| addCommand(new HelpCommand()); |
| addCommand(new BindCommand()); |
| addCommand(new UnbindCommand()); |
| addCommand(new RepairSeqNoCommand()); |
| addCommand(new DLCKCommand()); |
| addCommand(new SetDefaultACLCommand()); |
| addCommand(new SetStreamACLCommand()); |
| addCommand(new DeleteStreamACLCommand()); |
| } |
| |
| @Override |
| protected String getName() { |
| return "dlog_admin"; |
| } |
| } |
