kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

新增配置项，添加测试用例，修复在快速创建DataFile时，会创建出两个相同名字的DataFile

This commit is contained in:
ascrutae 2016-11-15 15:52:25 +08:00
parent 3a7e331aa0
commit c06b1fd0a7
8 changed files with 170 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/Main.java
View File

@ -30,7 +30,7 @@ import static com.a.eye.skywalking.storage.config.Config.RegistryCenter.PATH_PRE
public class Main {
private static final ILog logger = LogManager.getLogger(Main.class);
private static final String SERVER_REPORTER_NAME = "Storage Server";
private static final String SERVER_REPORTER_NAME = "DataConsumer Server";
static {
LogManager.setLogResolver(new Log4j2Resolver());

4
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/block/index/BlockFinder.java
View File

@ -24,10 +24,6 @@ public class BlockFinder {
index = l2Cache.find(timestamp);
}
if (logger.isDebugEnable()) {
logger.debug("Time stamp : {} is mapping with block Index {}.", timestamp, index);
}
return index;
}

10
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/config/Config.java
View File

@ -6,10 +6,16 @@ package com.a.eye.skywalking.storage.config;
public class Config {
public static class Server {
public static int PORT = 34000;
}
public static int CHANNEL_SIZE = 10;
public static int BUFFER_SIZE = 1000;
public static class DataConsumer {
public static int CHANNEL_SIZE = 10;
public static int BUFFER_SIZE = 1000;
public static int CONSUMER_SIZE = 5;
}

17
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/data/file/DataFile.java
View File

@ -1,5 +1,8 @@
package com.a.eye.skywalking.storage.data.file;
import com.a.eye.datacarrier.common.AtomicRangeInteger;
import com.a.eye.skywalking.health.report.HealthCollector;
import com.a.eye.skywalking.health.report.HeathReading;
import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.logging.api.LogManager;
import com.a.eye.skywalking.storage.config.Config;
@ -13,6 +16,9 @@ import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.atomic.AtomicInteger;
import static com.a.eye.skywalking.storage.util.PathResolver.getAbsolutePath;
@ -25,6 +31,7 @@ public class DataFile {
private String fileName;
private long currentOffset;
private DataFileOperator operator;
private static final AtomicRangeInteger DATA_FILE_NAME_SUFFIX = new AtomicRangeInteger(1000, 9999);
static {
File dataFileDir = new File(getAbsolutePath(Config.DataFile.PATH));
@ -34,7 +41,8 @@ public class DataFile {
}
public DataFile() {
this.fileName = System.currentTimeMillis() + "";
this.fileName = new SimpleDateFormat("yyyy_MM_dd_HH_mm_ss_SS").format(new Date()) + "_" + DATA_FILE_NAME_SUFFIX
.getAndIncrement();
this.currentOffset = 0;
operator = new DataFileOperator();
createFile();
@ -61,6 +69,8 @@ public class DataFile {
if (logger.isDebugEnable()) {
logger.debug("Create an new data file[{}].", fileName);
}
HealthCollector.getCurrentHeathReading("DataFile")
.updateData(HeathReading.INFO, "Create an new data " + "file.");
} catch (IOException e) {
logger.error("Failed to create data file.", e);
throw new DataFileOperatorCreateFailedException("Failed to create data file", e);
@ -96,7 +106,7 @@ public class DataFile {
}
}
public void close(){
public void close() {
operator.close();
}
@ -107,7 +117,8 @@ public class DataFile {
operator.getReader().read(data, 0, length);
return data;
} catch (IOException e) {
throw new SpanDataReadFailedException("Failed to read dataFile[" + fileName + "], offset: " + offset + " " + "lenght: " + length, e);
throw new SpanDataReadFailedException(
"Failed to read dataFile[" + fileName + "], offset: " + offset + " " + "lenght: " + length, e);
}
}

4
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/listener/StorageListener.java
View File

@ -20,8 +20,8 @@ public class StorageListener implements SpanStorageListener {
private DataCarrier<SpanData> spanDataDataCarrier;
public StorageListener() {
spanDataDataCarrier = new DataCarrier<>(Config.Server.CHANNEL_SIZE, Config.Server.BUFFER_SIZE);
spanDataDataCarrier.consume(new SpanDataConsumer(), 5, true);
spanDataDataCarrier = new DataCarrier<>(Config.DataConsumer.CHANNEL_SIZE, Config.DataConsumer.BUFFER_SIZE);
spanDataDataCarrier.consume(new SpanDataConsumer(), Config.DataConsumer.CONSUMER_SIZE, true);
}
@Override

9
skywalking-storage-center/skywalking-storage/src/main/resources/config.properties
View File

@ -1,6 +1,15 @@
# the port which storage server listening
server.port=34000
#
# the size of channel which storage span data
#dataconsumer.channel_size = 10
#
# the buffer size for each channel
#dataconsumer.buffer_size = 1000
#
# the size of data consumer
#dataconsumer.consumer_size = 5
#
# the path that storage block index
#blockindex.path=/block-index
#

78
skywalking-storage-center/skywalking-storage/src/test/java/StorageClient.java Normal file
View File

@ -0,0 +1,78 @@
import com.a.eye.skywalking.network.dependencies.io.grpc.ManagedChannel;
import com.a.eye.skywalking.network.dependencies.io.grpc.ManagedChannelBuilder;
import com.a.eye.skywalking.network.dependencies.io.grpc.stub.StreamObserver;
import com.a.eye.skywalking.network.grpc.AckSpan;
import com.a.eye.skywalking.network.grpc.RequestSpan;
import com.a.eye.skywalking.network.grpc.SendResult;
import com.a.eye.skywalking.network.grpc.SpanStorageServiceGrpc;
import static com.a.eye.skywalking.network.grpc.SpanStorageServiceGrpc.newStub;
public class StorageClient {
private static ManagedChannel channel =
ManagedChannelBuilder.forAddress("127.0.0.1", 34000).usePlaintext(true).build();
private static SpanStorageServiceGrpc.SpanStorageServiceStub spanStorageServiceStub = newStub(channel);
private static StreamObserver<AckSpan> ackSpanStreamObserver =
spanStorageServiceStub.storageACKSpan(new StreamObserver<SendResult>() {
@Override
public void onNext(SendResult sendResult) {
System.out.println(sendResult.getResult());
}
@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
}
@Override
public void onCompleted() {
System.out.println("Success!!");
}
});
private static StreamObserver<RequestSpan> requestSpanStreamObserver =
spanStorageServiceStub.storageRequestSpan(new StreamObserver<SendResult>() {
@Override
public void onNext(SendResult sendResult) {
System.out.println(sendResult.getResult());
}
@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
}
@Override
public void onCompleted() {
System.out.println("Success!!");
}
});
public static void main(String[] args) throws InterruptedException {
RequestSpan requestSpan =
RequestSpan.newBuilder().setSpanType(1).setAddress("127.0.0.1").setApplicationId("1").setCallType("1")
.setLevelId(0).setProcessNo("19287").setStartDate(System.currentTimeMillis())
.setTraceId("1.0Final.1478661327960.8504828.2277.53.3").setUserId("1")
.setViewPointId("http://localhost:8080/wwww/test/helloWorld").build();
AckSpan ackSpan =
AckSpan.newBuilder().setLevelId(0).setCost(10).setTraceId("1.0Final.1478661327960.8504828.2277.53.3")
.setStatusCode(0).setViewpointId("http://localhost:8080/wwww/test/helloWorld").build();
for (int i = 0; i < 100000; i++) {
requestSpanStreamObserver.onNext(requestSpan);
ackSpanStreamObserver.onNext(ackSpan);
Thread.sleep(100);
}
ackSpanStreamObserver.onCompleted();
requestSpanStreamObserver.onCompleted();
Thread.sleep(10000);
}
}

58
skywalking-storage-center/skywalking-storage/src/test/java/com/a/eye/skywalking/storage/data/file/DataFileWriterTest.java Normal file
View File

@ -0,0 +1,58 @@
package com.a.eye.skywalking.storage.data.file;
import com.a.eye.skywalking.network.grpc.RequestSpan;
import com.a.eye.skywalking.storage.config.Config;
import com.a.eye.skywalking.storage.data.spandata.RequestSpanData;
import com.a.eye.skywalking.storage.data.spandata.SpanData;
import com.a.eye.skywalking.storage.util.PathResolver;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.List;
import static org.junit.Assert.assertEquals;
/**
* Created by xin on 2016/11/15.
*/
public class DataFileWriterTest {
private DataFileWriter writer;
@Before
public void setUp() {
Config.DataFile.PATH = "/tmp";
Config.DataFile.SIZE = 10;
writer = new DataFileWriter();
}
@Test
public void testConvertFile() throws Exception {
List<SpanData> spanData = new ArrayList<>();
spanData.add(new RequestSpanData(
RequestSpan.newBuilder().setTraceId("test-traceId").setStartDate(System.currentTimeMillis())
.setProcessNo("7777").setLevelId(10).setParentLevel("0.0.0").setAddress("127.0.0.1").build()));
writer.write(spanData);
writer.write(spanData);
File dir = new File(PathResolver.getAbsolutePath(Config.DataFile.PATH));
assertEquals(2, dir.listFiles().length);
}
@After
public void tearUp() throws IOException {
File dir = new File(PathResolver.getAbsolutePath(Config.DataFile.PATH));
for (File file : dir.listFiles()) {
file.delete();
}
dir.delete();
}
}