Talk about Storm reportError

sequence

This article focuses on storm reportError

IErrorReporter

Storm – 2.0.0 / storm/org – client/SRC/JVM/apache/storm/task/IErrorReporter Java

public interface IErrorReporter {
    void reportError(Throwable error);
}
Copy the code

The ISpoutOutputCollector, IOutputCollector, and IBasicOutputCollector interfaces inherit the IErrorReporter interface

ISpoutOutputCollector

Storm – core / 1.2.2 / storm – core – 1.2.2 – sources. The jar! /org/apache/storm/spout/ISpoutOutputCollector.java

public interface ISpoutOutputCollector extends IErrorReporter{
    /**
        Returns the task ids that received the tuples.
    */
    List<Integer> emit(String streamId, List<Object> tuple, Object messageId);
    void emitDirect(int taskId, String streamId, List<Object> tuple, Object messageId);
    long getPendingCount();
}
Copy the code

ISpoutOutputCollector implements SpoutOutputCollector, SpoutOutputCollectorImpl, etc

IOutputCollector

Storm – 2.0.0 / storm/org – client/SRC/JVM/apache/storm/task/IOutputCollector Java

public interface IOutputCollector extends IErrorReporter {
    /**
     * Returns the task ids that received the tuples.
     */
    List<Integer> emit(String streamId, Collection<Tuple> anchors, List<Object> tuple);

    void emitDirect(int taskId, String streamId, Collection<Tuple> anchors, List<Object> tuple);

    void ack(Tuple input);

    void fail(Tuple input);

    void resetTimeout(Tuple input);

    void flush();
}
Copy the code

The implementation classes of IOutputCollector include OutputCollector, BoltOutputCollectorImpl, etc

IBasicOutputCollector

Storm – 2.0.0 / storm/org – client/SRC/JVM/apache/storm/topology/IBasicOutputCollector Java

public interface IBasicOutputCollector extends IErrorReporter { List<Integer> emit(String streamId, List<Object> tuple);  void emitDirect(int taskId, String streamId, List<Object> tuple); void resetTimeout(Tuple tuple); }Copy the code

The IBasicOutputCollector implementation class has BasicOutputCollector

reportError

SpoutOutputCollectorImpl.reportError

Storm – 2.0.0 / storm/org – client/SRC/JVM/apache/storm/executor/spout/SpoutOutputCollectorImpl. Java

    @Override
    public void reportError(Throwable error) {
        executor.getErrorReportingMetrics().incrReportedErrorCount();
        executor.getReportError().report(error);
    }
Copy the code

BoltOutputCollectorImpl.reportError

Storm – 2.0.0 / storm/org – client/SRC/JVM/apache/storm/executor/bolt/BoltOutputCollectorImpl. Java

    @Override
    public void reportError(Throwable error) {
        executor.getErrorReportingMetrics().incrReportedErrorCount();
        executor.getReportError().report(error);
    }
Copy the code

SpoutOutputCollectorImpl and BoltOutputCollectorImpl, reportError, both call executor.getrePorterror ().report(error);

ReportError.report

Storm – 2.0.0 / storm/org – client/SRC/JVM/apache/storm/executor/error/ReportError Java

public class ReportError implements IReportError {

    private static final Logger LOG = LoggerFactory.getLogger(ReportError.class);

    private final Map<String, Object> topoConf;
    private final IStormClusterState stormClusterState;
    private final String stormId;
    private final String componentId;
    private final WorkerTopologyContext workerTopologyContext;

    private int maxPerInterval;
    private int errorIntervalSecs;
    private AtomicInteger intervalStartTime;
    private AtomicInteger intervalErrors;

    public ReportError(Map<String, Object> topoConf, IStormClusterState stormClusterState, String stormId, String componentId,
                       WorkerTopologyContext workerTopologyContext) {
        this.topoConf = topoConf;
        this.stormClusterState = stormClusterState;
        this.stormId = stormId;
        this.componentId = componentId;
        this.workerTopologyContext = workerTopologyContext;
        this.errorIntervalSecs = ObjectReader.getInt(topoConf.get(Config.TOPOLOGY_ERROR_THROTTLE_INTERVAL_SECS));
        this.maxPerInterval = ObjectReader.getInt(topoConf.get(Config.TOPOLOGY_MAX_ERROR_REPORT_PER_INTERVAL));
        this.intervalStartTime = new AtomicInteger(Time.currentTimeSecs());
        this.intervalErrors = new AtomicInteger(0);
    }

    @Override
    public void report(Throwable error) {
        LOG.error("Error", error);
        if (Time.deltaSecs(intervalStartTime.get()) > errorIntervalSecs) {
            intervalErrors.set(0);
            intervalStartTime.set(Time.currentTimeSecs());
        }
        if(intervalErrors.incrementAndGet() <= maxPerInterval) { try { stormClusterState.reportError(stormId, componentId, Utils.hostname(), workerTopologyContext.getThisWorkerPort().longValue(), error); } catch (UnknownHostException e) { throw Utils.wrapInRuntime(e); }}}}Copy the code

Can see here to determine whether the interval first need to reset, and then determine whether the error than the interval of the largest number, more than words, call the stormClusterState. ReportError written to storage, such as zk

StormClusterStateImpl.reportError

Storm – 2.0.0 / storm/org – client/SRC/JVM/apache/storm/cluster/StormClusterStateImpl Java

    @Override
    public void reportError(String stormId, String componentId, String node, Long port, Throwable error) {
        String path = ClusterUtils.errorPath(stormId, componentId);
        String lastErrorPath = ClusterUtils.lastErrorPath(stormId, componentId);
        ErrorInfo errorInfo = new ErrorInfo(ClusterUtils.stringifyError(error), Time.currentTimeSecs());
        errorInfo.set_host(node);
        errorInfo.set_port(port.intValue());
        byte[] serData = Utils.serialize(errorInfo);
        stateStorage.mkdirs(path, defaultAcls);
        stateStorage.create_sequential(path + ClusterUtils.ZK_SEPERATOR + "e", serData, defaultAcls);
        stateStorage.set_data(lastErrorPath, serData, defaultAcls);
        List<String> childrens = stateStorage.get_children(path, false);

        Collections.sort(childrens, new Comparator<String>() {
            public int compare(String arg0, String arg1) {
                returnLong.compare(Long.parseLong(arg0.substring(1)), Long.parseLong(arg1.substring(1))); }});while (childrens.size() > 10) {
            String znodePath = path + ClusterUtils.ZK_SEPERATOR + childrens.remove(0);
            try {
                stateStorage.delete_node(znodePath);
            } catch (Exception e) {
                if (Utils.exceptionCauseIsInstanceOf(KeeperException.NoNodeException.class, e)) {
                    // if the node is already deleted, do nothing
                    LOG.warn("Could not find the znode: {}", znodePath);
                } else{ throw e; }}}}Copy the code

Here use ClusterUtils. ErrorPath (stormId componentId) write directory, again through the ClusterUtils. LastErrorPath (stormId componentId) access to the path of the writing
As ZK is not suitable for storing large amounts of data, if Childrens exceeds 10, the redundant nodes will be deleted in ascending order based on substring(1)

ClusterUtils.errorPath

Storm – 2.0.0 / storm/org – client/SRC/JVM/apache/storm/cluster/ClusterUtils Java

    public static final String ZK_SEPERATOR = "/";

    public static final String ERRORS_ROOT = "errors";

    public static final String ERRORS_SUBTREE = ZK_SEPERATOR + ERRORS_ROOT;

    public static String errorPath(String stormId, String componentId) {
        try {
            return errorStormRoot(stormId) + ZK_SEPERATOR + URLEncoder.encode(componentId, "UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw Utils.wrapInRuntime(e);
        }
    }

    public static String lastErrorPath(String stormId, String componentId) {
        return errorPath(stormId, componentId) + "-last-error";
    }

    public static String errorStormRoot(String stormId) {
        return ERRORS_SUBTREE + ZK_SEPERATOR + stormId;
    }
Copy the code

The path of errorPath is /errors/{stormId}/{componentId}. In this directory, the EPHEMERAL_SEQUENTIAL node starting with E is created. The error message is appended to this directory first, and then the old node is deleted if there are more than 10
The path of lastErrorPath is /errors/{stormId}/{componentId}-last-error, which stores the last error of this componentId

ZkCli view

[zk: localhost:2181(CONNECTED) 21] ls /storm/errors
[DRPCStateQuery-1-1540185943, reportErrorDemo-1-1540260375]
[zk: localhost:2181(CONNECTED) 22] ls /storm/errors/reportErrorDemo-1-1540260375
[print.print-last-error]
[zk: localhost:2181(CONNECTED) 23] ls /storm/errors/reportErrorDemo-1-1540260375/print
[e0000000291, e0000000290, e0000000295, e0000000294, e0000000293, e0000000292, e0000000299, e0000000298, e0000000297, e0000000296]
[zk: localhost:2181(CONNECTED) 24] ls /storm/errors/reportErrorDemo-1-1540260375/print/e0000000299
[]
[zk: localhost:2181(CONNECTED) 25] ls /storm/errors/reportErrorDemo-1-1540260375/print-last-error
[]
Copy the code

storm-ui

The curl -i http://192.168.99.100:8080/api/v1/topology/reportErrorDemo-1-1540260375? sys=false

Copy the code

Storm-ui requests the interface above to retrieve the topology data, where spout or Bolt contains lastError, showing the latest error message

StormApiResource

Storm – 2.0.0 / storm – webapp/SRC/main/Java/org/apache/storm/daemon/UI/resources/StormApiResource. Java

    /**
     * /api/v1/topology -> topo.
     */
    @GET
    @Path("/topology/{id}")
    @AuthNimbusOp(value = "getTopology", needsTopoId = true)
    @Produces("application/json")
    public Response getTopology(@PathParam("id") String id,
                                @DefaultValue(":all-time") @QueryParam("window") String window,
                                @QueryParam("sys") boolean sys,
                                @QueryParam(callbackParameterName) String callback) throws TException {
        topologyPageRequestMeter.mark();
        try (NimbusClient nimbusClient = NimbusClient.getConfiguredClient(config)) {
            returnUIHelpers.makeStandardResponse( UIHelpers.getTopologySummary( nimbusClient.getClient().getTopologyPageInfo(id, window, sys), window, config, servletRequest.getRemoteUser() ), callback ); }}Copy the code

The nimbusClient.getClient().gettopologypageInfo (ID, window, sys) method is called

Nimbus.getTopologyPageInfo

Storm – 2.0.0 / storm – server/SRC/main/Java/org/apache/storm/daemon/nimbus nimbus. Java

    @Override
    public TopologyPageInfo getTopologyPageInfo(String topoId, String window, boolean includeSys)
        throws NotAliveException, AuthorizationException, TException {
        try {
            getTopologyPageInfoCalls.mark();
            CommonTopoInfo common = getCommonTopoInfo(topoId, "getTopologyPageInfo");
            String topoName = common.topoName;
            IStormClusterState state = stormClusterState;
            int launchTimeSecs = common.launchTimeSecs;
            Assignment assignment = common.assignment;
            Map<List<Integer>, Map<String, Object>> beats = common.beats;
            Map<Integer, String> taskToComp = common.taskToComponent;
            StormTopology topology = common.topology;
            Map<String, Object> topoConf = Utils.merge(conf, common.topoConf);
            StormBase base = common.base;
            if (base == null) {
                throw new WrappedNotAliveException(topoId);
            }
            Map<WorkerSlot, WorkerResources> workerToResources = getWorkerResourcesForTopology(topoId);
            List<WorkerSummary> workerSummaries = null;
            Map<List<Long>, List<Object>> exec2NodePort = new HashMap<>();
            if(assignment ! = null) { Map<List<Long>, NodeInfo>execToNodeInfo = assignment.get_executor_node_port();
                Map<String, String> nodeToHost = assignment.get_node_host();
                for (Entry<List<Long>, NodeInfo> entry : execToNodeInfo.entrySet()) {
                    NodeInfo ni = entry.getValue();
                    List<Object> nodePort = Arrays.asList(ni.get_node(), ni.get_port_iterator().next());
                    exec2NodePort.put(entry.getKey(), nodePort);
                }

                workerSummaries = StatsUtil.aggWorkerStats(topoId,
                                                           topoName,
                                                           taskToComp,
                                                           beats,
                                                           exec2NodePort,
                                                           nodeToHost,
                                                           workerToResources,
                                                           includeSys,
                                                           true); //this is the topology page, so we know the user is authorized
            }

            TopologyPageInfo topoPageInfo = StatsUtil.aggTopoExecsStats(topoId,
                                                                        exec2NodePort, taskToComp, beats, topology, window, includeSys, state); / /...return topoPageInfo;
        } catch (Exception e) {
            LOG.warn("Get topo page info exception. (topology id='{}')", topoId, e);
            if(e instanceof TException) { throw (TException) e; } throw new RuntimeException(e); }}Copy the code

Here call StatsUtil. To obtain TopologyPageInfo aggTopoExecsStats

StatsUtil.aggTopoExecsStats

Storm – 2.0.0 / storm – server/SRC/main/Java/org/apache/storm/stats/StatsUtil Java

    /**
     * aggregate topo executors stats.
     *
     * @param topologyId     topology id
     * @param exec2nodePort  executor -> host+port
     * @param task2component task -> component
     * @param beats          executor[start, end] -> executor heartbeat
     * @param topology       storm topology
     * @param window         the window to be aggregated
     * @param includeSys     whether to include system streams
     * @param clusterState   cluster state
     * @return TopologyPageInfo thrift structure
     */
    public static TopologyPageInfo aggTopoExecsStats(
        String topologyId, Map exec2nodePort, Map task2component, Map<List<Integer>, Map<String, Object>> beats,
        StormTopology topology, String window, boolean includeSys, IStormClusterState clusterState) {
        List<Map<String, Object>> beatList = extractDataFromHb(exec2nodePort, task2component, beats, includeSys, topology);
        Map<String, Object> topoStats = aggregateTopoStats(window, includeSys, beatList);
        return postAggregateTopoStats(task2component, exec2nodePort, topoStats, topologyId, clusterState);
    }
Copy the code

StatsUtil aggTopoExecsStats method finally call postAggregateTopoStats

StatsUtil.postAggregateTopoStats