src/main/java/org/apache/accumulo/testing/continuous/ContinuousIngest.java - accumulo-testing - Git at Google

 /*
  * 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.accumulo.testing.continuous;

 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.apache.accumulo.testing.TestProps.CI_INGEST_PAUSE_DURATION_MAX;
 import static org.apache.accumulo.testing.TestProps.CI_INGEST_PAUSE_DURATION_MIN;
 import static org.apache.accumulo.testing.TestProps.CI_INGEST_PAUSE_WAIT_MAX;
 import static org.apache.accumulo.testing.TestProps.CI_INGEST_PAUSE_WAIT_MIN;

 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Properties;
 import java.util.Random;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.zip.CRC32;
 import java.util.zip.Checksum;

 import org.apache.accumulo.core.client.AccumuloClient;
 import org.apache.accumulo.core.client.BatchWriter;
 import org.apache.accumulo.core.client.MutationsRejectedException;
 import org.apache.accumulo.core.client.TableNotFoundException;
 import org.apache.accumulo.core.data.Mutation;
 import org.apache.accumulo.core.security.ColumnVisibility;
 import org.apache.accumulo.core.util.FastFormat;
 import org.apache.accumulo.testing.TestProps;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.google.common.base.Preconditions;

 public class ContinuousIngest {

   private static final Logger log = LoggerFactory.getLogger(ContinuousIngest.class);

   private static final byte[] EMPTY_BYTES = new byte[0];

   private static List<ColumnVisibility> visibilities;
   private static long lastPauseNs;
   private static long pauseWaitSec;

   private static ColumnVisibility getVisibility(Random rand) {
     return visibilities.get(rand.nextInt(visibilities.size()));
   }

   private static boolean pauseEnabled(Properties props) {
     String value = props.getProperty(TestProps.CI_INGEST_PAUSE_ENABLED);
     return Boolean.parseBoolean(value);
   }

   private static int getPause(Properties props, Random rand, String minProp, String maxProp) {
     int min = Integer.parseInt(props.getProperty(minProp));
     int max = Integer.parseInt(props.getProperty(maxProp));
     Preconditions.checkState(max >= min && min > 0);
     if (max == min) {
       return min;
     }
     return (rand.nextInt(max - min) + min);
   }

   private static int getFlushEntries(Properties props) {
     return Integer.parseInt(props.getProperty(TestProps.CI_INGEST_FLUSH_ENTRIES, "1000000"));
   }

   private static void pauseCheck(Properties props, Random rand) throws InterruptedException {
     if (pauseEnabled(props)) {
       long elapsedNano = System.nanoTime() - lastPauseNs;
       if (elapsedNano > (TimeUnit.SECONDS.toNanos(pauseWaitSec))) {
         long pauseDurationSec = getPause(props, rand, CI_INGEST_PAUSE_DURATION_MIN,
             CI_INGEST_PAUSE_DURATION_MAX);
         log.info("PAUSING for " + pauseDurationSec + "s");
         Thread.sleep(TimeUnit.SECONDS.toMillis(pauseDurationSec));
         lastPauseNs = System.nanoTime();
         pauseWaitSec = getPause(props, rand, CI_INGEST_PAUSE_WAIT_MIN, CI_INGEST_PAUSE_WAIT_MAX);
         log.info("INGESTING for " + pauseWaitSec + "s");
       }
     }
   }

   public static void main(String[] args) throws Exception {

     try (ContinuousEnv env = new ContinuousEnv(args)) {

       String vis = env.getTestProperty(TestProps.CI_INGEST_VISIBILITIES);
       if (vis == null) {
         visibilities = Collections.singletonList(new ColumnVisibility());
       } else {
         visibilities = new ArrayList<>();
         for (String v : vis.split(",")) {
           visibilities.add(new ColumnVisibility(v.trim()));
         }
       }

       long rowMin = env.getRowMin();
       long rowMax = env.getRowMax();
       if (rowMin < 0 || rowMax < 0 || rowMax <= rowMin) {
         throw new IllegalArgumentException("bad min and max");
       }

       AccumuloClient client = env.getAccumuloClient();
       String tableName = env.getAccumuloTableName();
       if (!client.tableOperations().exists(tableName)) {
         throw new TableNotFoundException(null, tableName,
             "Consult the README and create the table before starting ingest.");
       }

       BatchWriter bw = client.createBatchWriter(tableName);

       Random r = new Random();

       byte[] ingestInstanceId = UUID.randomUUID().toString().getBytes(UTF_8);

       log.info(String.format("UUID %d %s", System.currentTimeMillis(),
           new String(ingestInstanceId, UTF_8)));

       long count = 0;
       final int flushInterval = getFlushEntries(env.getTestProperties());
       final int maxDepth = 25;

       // always want to point back to flushed data. This way the previous item should
       // always exist in accumulo when verifying data. To do this make insert N point
       // back to the row from insert (N - flushInterval). The array below is used to keep
       // track of this.
       long[] prevRows = new long[flushInterval];
       long[] firstRows = new long[flushInterval];
       int[] firstColFams = new int[flushInterval];
       int[] firstColQuals = new int[flushInterval];

       long lastFlushTime = System.currentTimeMillis();

       int maxColF = env.getMaxColF();
       int maxColQ = env.getMaxColQ();
       boolean checksum = Boolean.parseBoolean(env.getTestProperty(TestProps.CI_INGEST_CHECKSUM));
       long numEntries = Long.parseLong(env.getTestProperty(TestProps.CI_INGEST_CLIENT_ENTRIES));

       Properties testProps = env.getTestProperties();
       if (pauseEnabled(testProps)) {
         lastPauseNs = System.nanoTime();
         pauseWaitSec = getPause(testProps, r, CI_INGEST_PAUSE_WAIT_MIN, CI_INGEST_PAUSE_WAIT_MAX);
         log.info("PAUSING enabled");
         log.info("INGESTING for " + pauseWaitSec + "s");
       }

       out: while (true) {
         // generate first set of nodes
         ColumnVisibility cv = getVisibility(r);

         for (int index = 0; index < flushInterval; index++) {
           long rowLong = genLong(rowMin, rowMax, r);
           prevRows[index] = rowLong;
           firstRows[index] = rowLong;

           int cf = r.nextInt(maxColF);
           int cq = r.nextInt(maxColQ);

           firstColFams[index] = cf;
           firstColQuals[index] = cq;

           Mutation m = genMutation(rowLong, cf, cq, cv, ingestInstanceId, count, null, checksum);
           count++;
           bw.addMutation(m);
         }

         lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
         if (count >= numEntries)
           break out;

         // generate subsequent sets of nodes that link to previous set of nodes
         for (int depth = 1; depth < maxDepth; depth++) {
           for (int index = 0; index < flushInterval; index++) {
             long rowLong = genLong(rowMin, rowMax, r);
             byte[] prevRow = genRow(prevRows[index]);
             prevRows[index] = rowLong;
             Mutation m = genMutation(rowLong, r.nextInt(maxColF), r.nextInt(maxColQ), cv,
                 ingestInstanceId, count, prevRow, checksum);
             count++;
             bw.addMutation(m);
           }

           lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
           if (count >= numEntries)
             break out;
           pauseCheck(testProps, r);
         }

         // create one big linked list, this makes all of the first inserts point to something
         for (int index = 0; index < flushInterval - 1; index++) {
           Mutation m = genMutation(firstRows[index], firstColFams[index], firstColQuals[index], cv,
               ingestInstanceId, count, genRow(prevRows[index + 1]), checksum);
           count++;
           bw.addMutation(m);
         }
         lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
         if (count >= numEntries)
           break out;
         pauseCheck(testProps, r);
       }
       bw.close();
     }
   }

   private static long flush(BatchWriter bw, long count, final int flushInterval, long lastFlushTime)
       throws MutationsRejectedException {
     long t1 = System.currentTimeMillis();
     bw.flush();
     long t2 = System.currentTimeMillis();
     log.info(String.format("FLUSH %d %d %d %d %d", t2, (t2 - lastFlushTime), (t2 - t1), count,
         flushInterval));
     lastFlushTime = t2;
     return lastFlushTime;
   }

   public static Mutation genMutation(long rowLong, int cfInt, int cqInt, ColumnVisibility cv,
       byte[] ingestInstanceId, long count, byte[] prevRow, boolean checksum) {
     // Adler32 is supposed to be faster, but according to wikipedia is not
     // good for small data.... so used CRC32 instead
     CRC32 cksum = null;

     byte[] rowString = genRow(rowLong);

     byte[] cfString = genCol(cfInt);
     byte[] cqString = genCol(cqInt);

     if (checksum) {
       cksum = new CRC32();
       cksum.update(rowString);
       cksum.update(cfString);
       cksum.update(cqString);
       cksum.update(cv.getExpression());
     }

     Mutation m = new Mutation(rowString);

     m.put(cfString, cqString, cv, createValue(ingestInstanceId, count, prevRow, cksum));
     return m;
   }

   public static byte[] genCol(int cfInt) {
     return FastFormat.toZeroPaddedString(cfInt, 4, 16, EMPTY_BYTES);
   }

   public static long genLong(long min, long max, Random r) {
     return ((r.nextLong() & 0x7fffffffffffffffL) % (max - min)) + min;
   }

   static byte[] genRow(long min, long max, Random r) {
     return genRow(genLong(min, max, r));
   }

   public static byte[] genRow(long rowLong) {
     return FastFormat.toZeroPaddedString(rowLong, 16, 16, EMPTY_BYTES);
   }

   public static byte[] createValue(byte[] ingestInstanceId, long count, byte[] prevRow,
       Checksum cksum) {
     int dataLen = ingestInstanceId.length + 16 + (prevRow == null ? 0 : prevRow.length) + 3;
     if (cksum != null)
       dataLen += 8;
     byte[] val = new byte[dataLen];
     System.arraycopy(ingestInstanceId, 0, val, 0, ingestInstanceId.length);
     int index = ingestInstanceId.length;
     val[index++] = ':';
     int added = FastFormat.toZeroPaddedString(val, index, count, 16, 16, EMPTY_BYTES);
     if (added != 16)
       throw new RuntimeException(" " + added);
     index += 16;
     val[index++] = ':';
     if (prevRow != null) {
       System.arraycopy(prevRow, 0, val, index, prevRow.length);
       index += prevRow.length;
     }

     val[index++] = ':';

     if (cksum != null) {
       cksum.update(val, 0, index);
       cksum.getValue();
       FastFormat.toZeroPaddedString(val, index, cksum.getValue(), 8, 16, EMPTY_BYTES);
     }
     return val;
   }
 }
	/*
	* 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.accumulo.testing.continuous;

	import static java.nio.charset.StandardCharsets.UTF_8;
	import static org.apache.accumulo.testing.TestProps.CI_INGEST_PAUSE_DURATION_MAX;
	import static org.apache.accumulo.testing.TestProps.CI_INGEST_PAUSE_DURATION_MIN;
	import static org.apache.accumulo.testing.TestProps.CI_INGEST_PAUSE_WAIT_MAX;
	import static org.apache.accumulo.testing.TestProps.CI_INGEST_PAUSE_WAIT_MIN;

	import java.util.ArrayList;
	import java.util.Collections;
	import java.util.List;
	import java.util.Properties;
	import java.util.Random;
	import java.util.UUID;
	import java.util.concurrent.TimeUnit;
	import java.util.zip.CRC32;
	import java.util.zip.Checksum;

	import org.apache.accumulo.core.client.AccumuloClient;
	import org.apache.accumulo.core.client.BatchWriter;
	import org.apache.accumulo.core.client.MutationsRejectedException;
	import org.apache.accumulo.core.client.TableNotFoundException;
	import org.apache.accumulo.core.data.Mutation;
	import org.apache.accumulo.core.security.ColumnVisibility;
	import org.apache.accumulo.core.util.FastFormat;
	import org.apache.accumulo.testing.TestProps;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.google.common.base.Preconditions;

	public class ContinuousIngest {

	private static final Logger log = LoggerFactory.getLogger(ContinuousIngest.class);

	private static final byte[] EMPTY_BYTES = new byte[0];

	private static List<ColumnVisibility> visibilities;
	private static long lastPauseNs;
	private static long pauseWaitSec;

	private static ColumnVisibility getVisibility(Random rand) {
	return visibilities.get(rand.nextInt(visibilities.size()));
	}

	private static boolean pauseEnabled(Properties props) {
	String value = props.getProperty(TestProps.CI_INGEST_PAUSE_ENABLED);
	return Boolean.parseBoolean(value);
	}

	private static int getPause(Properties props, Random rand, String minProp, String maxProp) {
	int min = Integer.parseInt(props.getProperty(minProp));
	int max = Integer.parseInt(props.getProperty(maxProp));
	Preconditions.checkState(max >= min && min > 0);
	if (max == min) {
	return min;
	}
	return (rand.nextInt(max - min) + min);
	}

	private static int getFlushEntries(Properties props) {
	return Integer.parseInt(props.getProperty(TestProps.CI_INGEST_FLUSH_ENTRIES, "1000000"));
	}

	private static void pauseCheck(Properties props, Random rand) throws InterruptedException {
	if (pauseEnabled(props)) {
	long elapsedNano = System.nanoTime() - lastPauseNs;
	if (elapsedNano > (TimeUnit.SECONDS.toNanos(pauseWaitSec))) {
	long pauseDurationSec = getPause(props, rand, CI_INGEST_PAUSE_DURATION_MIN,
	CI_INGEST_PAUSE_DURATION_MAX);
	log.info("PAUSING for " + pauseDurationSec + "s");
	Thread.sleep(TimeUnit.SECONDS.toMillis(pauseDurationSec));
	lastPauseNs = System.nanoTime();
	pauseWaitSec = getPause(props, rand, CI_INGEST_PAUSE_WAIT_MIN, CI_INGEST_PAUSE_WAIT_MAX);
	log.info("INGESTING for " + pauseWaitSec + "s");
	}
	}
	}

	public static void main(String[] args) throws Exception {

	try (ContinuousEnv env = new ContinuousEnv(args)) {

	String vis = env.getTestProperty(TestProps.CI_INGEST_VISIBILITIES);
	if (vis == null) {
	visibilities = Collections.singletonList(new ColumnVisibility());
	} else {
	visibilities = new ArrayList<>();
	for (String v : vis.split(",")) {
	visibilities.add(new ColumnVisibility(v.trim()));
	}
	}

	long rowMin = env.getRowMin();
	long rowMax = env.getRowMax();
	if (rowMin < 0 \|\| rowMax < 0 \|\| rowMax <= rowMin) {
	throw new IllegalArgumentException("bad min and max");
	}

	AccumuloClient client = env.getAccumuloClient();
	String tableName = env.getAccumuloTableName();
	if (!client.tableOperations().exists(tableName)) {
	throw new TableNotFoundException(null, tableName,
	"Consult the README and create the table before starting ingest.");
	}

	BatchWriter bw = client.createBatchWriter(tableName);

	Random r = new Random();

	byte[] ingestInstanceId = UUID.randomUUID().toString().getBytes(UTF_8);

	log.info(String.format("UUID %d %s", System.currentTimeMillis(),
	new String(ingestInstanceId, UTF_8)));

	long count = 0;
	final int flushInterval = getFlushEntries(env.getTestProperties());
	final int maxDepth = 25;

	// always want to point back to flushed data. This way the previous item should
	// always exist in accumulo when verifying data. To do this make insert N point
	// back to the row from insert (N - flushInterval). The array below is used to keep
	// track of this.
	long[] prevRows = new long[flushInterval];
	long[] firstRows = new long[flushInterval];
	int[] firstColFams = new int[flushInterval];
	int[] firstColQuals = new int[flushInterval];

	long lastFlushTime = System.currentTimeMillis();

	int maxColF = env.getMaxColF();
	int maxColQ = env.getMaxColQ();
	boolean checksum = Boolean.parseBoolean(env.getTestProperty(TestProps.CI_INGEST_CHECKSUM));
	long numEntries = Long.parseLong(env.getTestProperty(TestProps.CI_INGEST_CLIENT_ENTRIES));

	Properties testProps = env.getTestProperties();
	if (pauseEnabled(testProps)) {
	lastPauseNs = System.nanoTime();
	pauseWaitSec = getPause(testProps, r, CI_INGEST_PAUSE_WAIT_MIN, CI_INGEST_PAUSE_WAIT_MAX);
	log.info("PAUSING enabled");
	log.info("INGESTING for " + pauseWaitSec + "s");
	}

	out: while (true) {
	// generate first set of nodes
	ColumnVisibility cv = getVisibility(r);

	for (int index = 0; index < flushInterval; index++) {
	long rowLong = genLong(rowMin, rowMax, r);
	prevRows[index] = rowLong;
	firstRows[index] = rowLong;

	int cf = r.nextInt(maxColF);
	int cq = r.nextInt(maxColQ);

	firstColFams[index] = cf;
	firstColQuals[index] = cq;

	Mutation m = genMutation(rowLong, cf, cq, cv, ingestInstanceId, count, null, checksum);
	count++;
	bw.addMutation(m);
	}

	lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
	if (count >= numEntries)
	break out;

	// generate subsequent sets of nodes that link to previous set of nodes
	for (int depth = 1; depth < maxDepth; depth++) {
	for (int index = 0; index < flushInterval; index++) {
	long rowLong = genLong(rowMin, rowMax, r);
	byte[] prevRow = genRow(prevRows[index]);
	prevRows[index] = rowLong;
	Mutation m = genMutation(rowLong, r.nextInt(maxColF), r.nextInt(maxColQ), cv,
	ingestInstanceId, count, prevRow, checksum);
	count++;
	bw.addMutation(m);
	}

	lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
	if (count >= numEntries)
	break out;
	pauseCheck(testProps, r);
	}

	// create one big linked list, this makes all of the first inserts point to something
	for (int index = 0; index < flushInterval - 1; index++) {
	Mutation m = genMutation(firstRows[index], firstColFams[index], firstColQuals[index], cv,
	ingestInstanceId, count, genRow(prevRows[index + 1]), checksum);
	count++;
	bw.addMutation(m);
	}
	lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
	if (count >= numEntries)
	break out;
	pauseCheck(testProps, r);
	}
	bw.close();
	}
	}

	private static long flush(BatchWriter bw, long count, final int flushInterval, long lastFlushTime)
	throws MutationsRejectedException {
	long t1 = System.currentTimeMillis();
	bw.flush();
	long t2 = System.currentTimeMillis();
	log.info(String.format("FLUSH %d %d %d %d %d", t2, (t2 - lastFlushTime), (t2 - t1), count,
	flushInterval));
	lastFlushTime = t2;
	return lastFlushTime;
	}

	public static Mutation genMutation(long rowLong, int cfInt, int cqInt, ColumnVisibility cv,
	byte[] ingestInstanceId, long count, byte[] prevRow, boolean checksum) {
	// Adler32 is supposed to be faster, but according to wikipedia is not
	// good for small data.... so used CRC32 instead
	CRC32 cksum = null;

	byte[] rowString = genRow(rowLong);

	byte[] cfString = genCol(cfInt);
	byte[] cqString = genCol(cqInt);

	if (checksum) {
	cksum = new CRC32();
	cksum.update(rowString);
	cksum.update(cfString);
	cksum.update(cqString);
	cksum.update(cv.getExpression());
	}

	Mutation m = new Mutation(rowString);

	m.put(cfString, cqString, cv, createValue(ingestInstanceId, count, prevRow, cksum));
	return m;
	}

	public static byte[] genCol(int cfInt) {
	return FastFormat.toZeroPaddedString(cfInt, 4, 16, EMPTY_BYTES);
	}

	public static long genLong(long min, long max, Random r) {
	return ((r.nextLong() & 0x7fffffffffffffffL) % (max - min)) + min;
	}

	static byte[] genRow(long min, long max, Random r) {
	return genRow(genLong(min, max, r));
	}

	public static byte[] genRow(long rowLong) {
	return FastFormat.toZeroPaddedString(rowLong, 16, 16, EMPTY_BYTES);
	}

	public static byte[] createValue(byte[] ingestInstanceId, long count, byte[] prevRow,
	Checksum cksum) {
	int dataLen = ingestInstanceId.length + 16 + (prevRow == null ? 0 : prevRow.length) + 3;
	if (cksum != null)
	dataLen += 8;
	byte[] val = new byte[dataLen];
	System.arraycopy(ingestInstanceId, 0, val, 0, ingestInstanceId.length);
	int index = ingestInstanceId.length;
	val[index++] = ':';
	int added = FastFormat.toZeroPaddedString(val, index, count, 16, 16, EMPTY_BYTES);
	if (added != 16)
	throw new RuntimeException(" " + added);
	index += 16;
	val[index++] = ':';
	if (prevRow != null) {
	System.arraycopy(prevRow, 0, val, index, prevRow.length);
	index += prevRow.length;
	}

	val[index++] = ':';

	if (cksum != null) {
	cksum.update(val, 0, index);
	cksum.getValue();
	FastFormat.toZeroPaddedString(val, index, cksum.getValue(), 8, 16, EMPTY_BYTES);
	}
	return val;
	}
	}