public class OAPServerBootstrap {
    public static void start() {
    // 初始化mode为init或者no-init,表示是否初始化[example:底层存储组件等]
        String mode = System.getProperty("mode");
        RunningMode.setMode(mode);
      // 初始化ApplicationConfigurationd的加载器和Module的加载管理器
        ApplicationConfigLoader configLoader = new ApplicationConfigLoader();
        ModuleManager manager = new ModuleManager();
        try {
        // 加载yml生成ApplicationConfiguration配置
        
            ApplicationConfiguration applicationConfiguration = configLoader.load();
            
            // 初始化模块 通过spi获取所有Module实现,基于yml配置加载spi中存在的相关实现
            manager.init(applicationConfiguration);

            manager.find(TelemetryModule.NAME)
                   .provider()
                   .getService(MetricsCreator.class)
                   .createGauge("uptime", "oap server start up time", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE)
                   // Set uptime to second
                   .setValue(System.currentTimeMillis() / 1000d);

            if (RunningMode.isInitMode()) {
                log.info("OAP starts up in init mode successfully, exit now...");
                System.exit(0);
            }
        } catch (Throwable t) {
            log.error(t.getMessage(), t);
            System.exit(1);
        }
    }
}

# CoreModuleProvider
public void prepare() {
    if (moduleConfig.isGRPCSslEnabled()) {
        grpcServer = new GRPCServer(moduleConfig.getGRPCHost(), moduleConfig.getGRPCPort(),
                                    moduleConfig.getGRPCSslCertChainPath(),
                                    moduleConfig.getGRPCSslKeyPath()
        );
    } else {
        grpcServer = new GRPCServer(moduleConfig.getGRPCHost(), moduleConfig.getGRPCPort());
    }
    if (moduleConfig.getMaxConcurrentCallsPerConnection() > 0) {
        grpcServer.setMaxConcurrentCallsPerConnection(moduleConfig.getMaxConcurrentCallsPerConnection());
    }
    if (moduleConfig.getMaxMessageSize() > 0) {
        grpcServer.setMaxMessageSize(moduleConfig.getMaxMessageSize());
    }
    if (moduleConfig.getGRPCThreadPoolQueueSize() > 0) {
        grpcServer.setThreadPoolQueueSize(moduleConfig.getGRPCThreadPoolQueueSize());
    }
    if (moduleConfig.getGRPCThreadPoolSize() > 0) {
        grpcServer.setThreadPoolSize(moduleConfig.getGRPCThreadPoolSize());
    }
    grpcServer.initialize();

}

    @Override
    public StreamObserver<SegmentObject> collect(StreamObserver<Commands> responseObserver) {
        return new StreamObserver<SegmentObject>() {
            @Override
            public void onNext(SegmentObject segment) {
                if (log.isDebugEnabled()) {
                    log.debug("received segment in streaming");
                }

                HistogramMetrics.Timer timer = histogram.createTimer();
                try {
                    segmentParserService.send(segment);
                } catch (Exception e) {
                    errorCounter.inc();
                    log.error(e.getMessage(), e);
                } finally {
                    timer.finish();
                }
            }

            @Override
            public void onError(Throwable throwable) {
                log.error(throwable.getMessage(), throwable);
                responseObserver.onCompleted();
            }

            @Override
            public void onCompleted() {
                responseObserver.onNext(Commands.newBuilder().build());
                responseObserver.onCompleted();
            }
        };
    }

@RequiredArgsConstructor
public class SegmentParserServiceImpl implements ISegmentParserService {
    private final ModuleManager moduleManager;
    private final AnalyzerModuleConfig config;
    @Setter
    private SegmentParserListenerManager listenerManager;

    @Override
    public void send(SegmentObject segment) {
        final TraceAnalyzer traceAnalyzer = new TraceAnalyzer(moduleManager, listenerManager, config);
        traceAnalyzer.doAnalysis(segment);
    }
}

public void doAnalysis(SegmentObject segmentObject) {
    if (segmentObject.getSpansList().size() == 0) {
        return;
    }
    // 创建span的listener
    createSpanListeners();
    // 通知segment监听，构建存储model
    notifySegmentListener(segmentObject);
    // 通知 first exit local的各个listen，构建存储model
    segmentObject.getSpansList().forEach(spanObject -> {
        if (spanObject.getSpanId() == 0) {
            notifyFirstListener(spanObject, segmentObject);
        }

        if (SpanType.Exit.equals(spanObject.getSpanType())) {
            notifyExitListener(spanObject, segmentObject);
        } else if (SpanType.Entry.equals(spanObject.getSpanType())) {
            notifyEntryListener(spanObject, segmentObject);
        } else if (SpanType.Local.equals(spanObject.getSpanType())) {
            notifyLocalListener(spanObject, segmentObject);
        } else {
            log.error("span type value was unexpected, span type name: {}", spanObject.getSpanType()
                                                                                      .name());
        }
    });
    // 触发存储动作
    notifyListenerToBuild();
}

private void notifyListenerToBuild() {
    analysisListeners.forEach(AnalysisListener::build);
}

// 去掉非核心代码
public void build() {
    segment.setEndpointId(endpointId);
    segment.setEndpointName(endpointName);
    // 存储核心
    sourceReceiver.receive(segment);
}

public void receive(ISource source) {
     dispatcherManager.forward(source);
}

public void forward(ISource source) {
    source.prepare();
    for (SourceDispatcher dispatcher : dispatchers) {
        dispatcher.dispatch(source);
    }   
}

public void dispatch(Segment source) {
    SegmentRecord segment = new SegmentRecord();
    segment.setSegmentId(source.getSegmentId());
    segment.setTraceId(source.getTraceId());
    segment.setServiceId(source.getServiceId());
    segment.setServiceInstanceId(source.getServiceInstanceId());
    segment.setEndpointName(source.getEndpointName());
    segment.setEndpointId(source.getEndpointId());
    segment.setStartTime(source.getStartTime());
    segment.setEndTime(source.getEndTime());
    segment.setLatency(source.getLatency());
    segment.setIsError(source.getIsError());
    segment.setDataBinary(source.getDataBinary());
    segment.setTimeBucket(source.getTimeBucket());
    segment.setVersion(source.getVersion());
    segment.setTagsRawData(source.getTags());
    segment.setTags(Tag.Util.toStringList(source.getTags()));

    RecordStreamProcessor.getInstance().in(segment);
}

public void in(Record record) {
    InsertRequest insertRequest = recordDAO.prepareBatchInsert(model, record);
    batchDAO.asynchronous(insertRequest);    
}

public void asynchronous(InsertRequest insertRequest) {
    if (bulkProcessor == null) {
        this.bulkProcessor = getClient().createBulkProcessor(bulkActions, flushInterval, concurrentRequests);
    }

    this.bulkProcessor.add((IndexRequest) insertRequest);
}

  source == Source.Agent

  #parse(UpstreamSegment, Source)