Response headers named metrics.* are fed straight into the fetcher metric registry, and a non-numeric one fails the fetch

[rzo1]

What happens

The okhttp protocol copies every response header into the response metadata, lowercased and with no prefix. FetcherBolt then iterates response.getMetadata().keySet("metrics.") and calls averagedMetrics.scope(name).update(Long.parseLong(value)). The prefix used to select internal timings is a legal header name prefix, so the server on the other end writes into that namespace. Long.parseLong is not guarded, so a header value that is not a number throws inside the fetch try block and the URL is recorded as FETCH_ERROR even though the response arrived intact. SimpleFetcherBolt has the same loop.

Where

core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java:770-778, core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java:476-484, headers ingested at core/src/main/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocol.java:434-445. The protocol's own timing is written at HttpProtocol.java:464. No configuration key is involved; the loop runs at defaults.

                    response.getMetadata().keySet("metrics.").stream()
                            .forEach(
                                    s ->
                                            averagedMetrics
                                                    .scope(s.substring(8))
                                                    .update(
                                                            Long.parseLong(
                                                                    response.getMetadata()
                                                                            .getFirstValue(s))));

                responsemetadata.addValue(key.toLowerCase(Locale.ROOT), value);

Why it matters

A page that is fetched without incident is reported as a fetch error, and the reason logged is a bare NumberFormatException, which is hard to connect to the response header that caused it. Beyond that, every distinct metrics.<name> header allocates a metric scope that is never evicted: the V2 path registers a histogram per name in a ConcurrentHashMap that is never cleaned up, and the V1 path delegates to Storm's own MultiReducedMetric, which keeps one entry per scope for the lifetime of the worker. A long crawl over a host that varies the header name grows that map for the lifetime of the worker and fills dashboards with series the operator did not define. Values under names the fetcher uses itself, such as the DNS timing, are also writable, except when the protocol overwrites them for that response.

Reproduction

Save as core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltMetricsHeaderTest.java.

/*
 * 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.stormcrawler.bolt;

import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
import static com.github.tomakehurst.wiremock.client.WireMock.get;
import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo;
import com.github.tomakehurst.wiremock.junit5.WireMockTest;
import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.utils.Utils;
import org.apache.stormcrawler.Constants;
import org.apache.stormcrawler.TestOutputCollector;
import org.apache.stormcrawler.TestUtil;
import org.apache.stormcrawler.protocol.ProtocolFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

/**
 * A response header called metrics.* is server data. It must not decide whether the page counts as
 * fetched.
 */
@WireMockTest
class FetcherBoltMetricsHeaderTest {

    private BaseRichBolt bolt;

    @BeforeEach
    void setUp() {
        bolt = new FetcherBolt();
    }

    @AfterEach
    void cleanup() {
        bolt.cleanup();
    }

    @Test
    void nonNumericMetricsHeaderDoesNotFailTheFetch(WireMockRuntimeInfo wmRuntimeInfo)
            throws Exception {
        stubFor(
                get(urlEqualTo("/page"))
                        .willReturn(
                                aResponse()
                                        .withStatus(200)
                                        .withHeader("metrics.example", "not-a-number")
                                        .withBody("hello")));

        Field instance = ProtocolFactory.class.getDeclaredField("single_instance");
        instance.setAccessible(true);
        instance.set(null, null);

        TestOutputCollector output = new TestOutputCollector();
        Map<String, Object> config = new HashMap<>();
        config.put("http.agent.name", "this_is_only_a_test");
        bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output));

        Tuple tuple = mock(Tuple.class);
        when(tuple.getSourceComponent()).thenReturn("source");
        when(tuple.getStringByField("url"))
                .thenReturn("http://localhost:" + wmRuntimeInfo.getHttpPort() + "/page");
        when(tuple.getValueByField("metadata")).thenReturn(null);
        bolt.execute(tuple);

        await().atMost(30, TimeUnit.SECONDS)
                .until(
                        () ->
                                output.getEmitted(Utils.DEFAULT_STREAM_ID).size() > 0
                                        || output.getEmitted(Constants.StatusStreamName).size() > 0);

        List<List<Object>> statusTuples = output.getEmitted(Constants.StatusStreamName);
        if (!statusTuples.isEmpty()) {
            Assertions.fail("emitted on the status stream: " + statusTuples.get(0).get(2));
        }
        Assertions.assertEquals(1, output.getEmitted(Utils.DEFAULT_STREAM_ID).size());
    }
}

Run it:

mvn -pl core test -Dtest=FetcherBoltMetricsHeaderTest

It uses the WireMock setup already used by FetcherBoltTest, serves a 200 response with a metrics.example: not-a-number header, and asserts the page is emitted on the default stream. It fails on main.

INFO org.apache.stormcrawler.bolt.FetcherBolt - Exception while fetching http://localhost:52373/page -> java.lang.NumberFormatException
[ERROR] Tests run: 1, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 1.732 s <<< FAILURE! -- in org.apache.stormcrawler.bolt.FetcherBoltMetricsHeaderTest
org.opentest4j.AssertionFailedError: emitted on the status stream: FETCH_ERROR

Suggested fix

Keep server data out of the metric namespace. In HttpProtocol.getProtocolOutput, skip or rename incoming headers that start with metrics. when building the response metadata, since the protocol adds its own entry afterwards. In FetcherBolt.FetcherThread.run and the matching loop in SimpleFetcherBolt.execute, wrap the Long.parseLong per key in a try/catch that logs and skips the value, so a malformed entry cannot change the status of the URL. If a use case for protocol supplied metrics from other implementations exists, restrict the loop to a fixed set of known names rather than a prefix.