Given our discussion in #24, and previous discussions on Kafka spouts, HBase indexing, etc, we should think about reorganizing the project so that we have a core SDK and external SDK(s).

I was thinking something like

root      pom.xml     |-> crawler-core         pom.xml     |-> crawler-external         pom.xml

The external sub-project would include things that depend on external technologies and libraries.

• [x ] Bug report. If you’ve found a bug, please include a test if you can, it makes it a lot easier to fix things. Use the label 'bug' on the issue.

I am getting the following :-

16:33:27.858 [Thread-35-spout-executor[12, 12]] ERROR c.d.s.e.p.AbstractSpout - Can't connect to ElasticSearch
java.lang.IllegalArgumentException: hosts must not be null nor empty
        at org.elasticsearch.client.RestClient.builder(RestClient.java:173) ~[stormcrawler-1.0-SNAPSHOT.jar:?]
        at com.digitalpebble.stormcrawler.elasticsearch.ElasticSearchConnection.getClient(ElasticSearchConnection.java:117) ~[stormcrawler-1.0-SNAPSHOT.jar:?]

ElasticSearch is up and running. I cannot find a 'hosts:' parameter anywhere and 'es-conf.yaml' parameters are set to localhost, or http://locahost:9200

This PR changes the behaviour of the SQL StatusUpdaterBolt by batching the INSERTs, i.e. the discovered URLs which are far more frequent than updates.

@cruftex plz let me know what you think. we can make it SQL agnostic in a separate step.

Hello @jnioche,

I extracted the code for initializing classes from string and configuring them from my fork https://github.com/FelixEngl/storm-crawler/tree/local_version. (This branch contains all fixes/changes that i made. I'll try to extract PR after PR until both are on a same level.)

I don't know how far you got with #937, but I this one has a more reasonable size with the trade-off that it introduces some new warnings due to missing @Contract, @NotNull and @Nullable annotations in various sub-classes and sub-interfaces. These warnings will be fixed by either #937 or a PR in the future.

Best Regards

Felix

Signed-off-by: Felix Engl [email protected]

Developer Certificate of Origin

Developer Certificate of Origin
Version 1.1

Copyright (C) 2004, 2006 The Linux Foundation and its contributors.
1 Letterman Drive
Suite D4700
San Francisco, CA, 94129

Everyone is permitted to copy and distribute verbatim copies of this
license document, but changing it is not allowed.


Developer's Certificate of Origin 1.1

By making a contribution to this project, I certify that:

(a) The contribution was created in whole or in part by me and I
    have the right to submit it under the open source license
    indicated in the file; or

(b) The contribution is based upon previous work that, to the best
    of my knowledge, is covered under an appropriate open source
    license and I have the right under that license to submit that
    work with modifications, whether created in whole or in part
    by me, under the same open source license (unless I am
    permitted to submit under a different license), as indicated
    in the file; or

(c) The contribution was provided directly to me by some other
    person who certified (a), (b) or (c) and I have not modified
    it.

(d) I understand and agree that this project and the contribution
    are public and that a record of the contribution (including all
    personal information I submit with it, including my sign-off) is
    maintained indefinitely and may be redistributed consistent with
    this project or the open source license(s) involved.

Currently the Solr storage modules send hard commits to Solr.

I think this is not the good way to proceed. These commits cause too much pressure on Solr. See the following screenshot.

Instead, these modules should let Solr commit the documents with 'autoCommit' and 'autoSoftCommit' sections in the 'solrconfig' file.

Implements #688 and fixes #684

This adds a ScrollSpout for ES as well as a mechanism to the AbstractStatusUpdaterBolt so that it stores a tuple without modifying any of its content.

The following Flux illustrates its use

name: "reindexer"

includes:
    - resource: true
      file: "/crawler-default.yaml"
      override: false

    - resource: false
      file: "crawler-conf.yaml"
      override: true

    - resource: false
      file: "es-conf.yaml"
      override: true

config:
  es.status2.addresses: "localhost"
  es.status2.index.name: "status2"
  es.status2.doc.type: "status"
  es.status2.routing: true
  es.status2.routing.fieldname: "key"
  es.status2.bulkActions: 500
  es.status2.flushInterval: "1s"
  es.status2.concurrentRequests: 5
  es.status2.settings:
    cluster.name: "elasticsearch"
  topology.max.spout.pending: 5000
  topology.workers: 1

spouts:
  - id: "spout"
    className: "com.digitalpebble.stormcrawler.elasticsearch.persistence.ScrollSpout"
    parallelism: 10

bolts:
  - id: "status"
    className: "com.digitalpebble.stormcrawler.elasticsearch.persistence.StatusUpdaterBolt"
    parallelism: 4
    constructorArgs:
      - "status2"

streams:
  - from: "spout"
    to: "status"
    grouping:
      streamId: "status"
      type: CUSTOM
      customClass:
        className: "com.digitalpebble.stormcrawler.util.URLStreamGrouping"
        constructorArgs:
          - "byDomain"

The target index (here 'status2)' has to be initialised just like the source one. It can of course live on a separate cluster.

It is a good idea to set "refresh_interval": "-1" in the configuration of the target index to speed up the writes. This can be set to any value afterwards when crawling with the new index.

Copying the content of a status index can be useful e.g. when changing the number of shards or the way the documents are assigned to them - for instance if using domains with a different version of crawler-commons, see #684

Based on #257, uses ES 2.3.1

The only limitation is that the node client does not work for now, but the transport one does.

@w0mbat thanks for your work on #257 - I've built a new branch but yours was a great starting point. Any chance you could give this one a try?

This address the https://github.com/DigitalPebble/storm-crawler/issues/117 issue at some extent this also encapsulate in a ParseResult the parsing data/metadata of a URL. Each URL has only one ParseResult which at the same time has at least 1 ParseData instance to hold the data of the "parent" URL in case subdocuments are extracted in any ParseFilter the URLs and ParseData of each subdocument is added to the parent ParseResult. So the ParserBolt emits each ParseData container in the ParseResult as a tuple. So at the very least 1 ParseData, or any number if subdocuments are extracted, also this implies that there is no difference between the extracted information about the parent URL or any subdocument each ParseData gets emitted as a tuple.

I made some simple graph to try to explain this :) I hope it makes things a little more clearer than my explanation.

See #38

This adds a SiteMapParserBolt and a test class. It also reorganizes the test resources and makes the ParsingTester more generic.

As discussed, this Bolt uses a non-default stream 'status' to output the newly discovered URLs and the default stream for documents that are not marked as being sitemaps.

Could you please review this? Thanks!

OkHttp's ConnectionPool by default "holds up to 5 idle connections which will be evicted after 5 minutes of inactivity." A pool of this size is suitable for site crawls but not for larger crawls over many different sites.

Note: in the current version (4.9.2) the connection pool is implemented as a linked queue and searching for a pooled connection does not scale up. In order to scale beyond pool sizes exceeding 1000 a set of clients must be used each with its own connection pool.

Notes:

• so far, only partially tested: need to increase the pool size and run a test crawl to measure the impact
• proxied connections are unchanged, that is, for every fetch a client is created anew and no connection pool is used. Depending on the proxy manager, it could make sense to define a connection pool ahead and pass it to the client builder. Since proxy information is included in the okhttp address (stored in the connection pool) it should be possible to pool proxied connections.
• the documentation could be moved from crawler-default.yaml to the wiki

Thanks for contributing to StormCrawler, your efforts are appreciated!

Developer Certificate of Origin

By contributing to StormCrawler, you accept and agree to the following terms and conditions (the Developer Certificate of Origin) for your present and future contributions submitted to StormCrawler. Please refer to the Developer Certificate of Origin section in CONTRIBUTING.md for details.

Developer Certificate of Origin
Version 1.1

Copyright (C) 2004, 2006 The Linux Foundation and its contributors.
1 Letterman Drive
Suite D4700
San Francisco, CA, 94129

Everyone is permitted to copy and distribute verbatim copies of this
license document, but changing it is not allowed.


Developer's Certificate of Origin 1.1

By making a contribution to this project, I certify that:

(a) The contribution was created in whole or in part by me and I
    have the right to submit it under the open source license
    indicated in the file; or

(b) The contribution is based upon previous work that, to the best
    of my knowledge, is covered under an appropriate open source
    license and I have the right under that license to submit that
    work with modifications, whether created in whole or in part
    by me, under the same open source license (unless I am
    permitted to submit under a different license), as indicated
    in the file; or

(c) The contribution was provided directly to me by some other
    person who certified (a), (b) or (c) and I have not modified
    it.

(d) I understand and agree that this project and the contribution
    are public and that a record of the contribution (including all
    personal information I submit with it, including my sign-off) is
    maintained indefinitely and may be redistributed consistent with
    this project or the open source license(s) involved.

Before opening a PR, please check that:

• You've applied the formatting used by the project with mvn java-formatter:format
• You've squashed your commits into a single one
• You've described what the PR does or at least point to a related issue
• You've signed-ff your commits with 'git commit -s'

Thanks!

The pom files in the archetype should be update to java 11 version rather than jdk 8 https://github.com/DigitalPebble/storm-crawler/blob/master/external/elasticsearch/archetype/src/main/resources/archetype-resources/pom.xml

Hi @jnioche !

Thanks again for all your work ! Now, let me expose you our fetcher thread issue.

Resume

Our cluster have 6 worker nodes. We are fetching more than 3 million URLs per day with our topology. It is deployed on 16 worker slots and use 16 fetchers, one by worker slot.

OkClient.HttpProtocol

The worst issue was spotted with the OkClient.HttpProtocol. Sometime, one of the worker nodes step up to 100% CPU usage. For example, the worker 5 in this case:

On StromCrawler board, we can see the fetcher count increase up to 50 (our fetcher limit) :

Worst, in another case, all the topologies are impacted :

All fetchers are impacted, and the topology is running slowly. The only way to fix the problem, is to kill and redeploy the topology. On kill phase, the log confirms some blocking thread:

2022-05-30 06:37:06.557 O.A.S.D.W.WORKER SHUTDOWNHOOK-SHUTDOWNFUNC [INFO] SHUTTING DOWN EXECUTORS ... 2022-05-30 06:37:07.028 O.A.S.E.EXECUTORSHUTDOWN SHUTDOWNHOOK-SHUTDOWNFUNC [INFO] SHUTTING DOWN EXECUTOR FETCHER:[30, 30] 2022-05-30 06:37:07.077 C.D.S.B.FETCHERBOLT THREAD-21-FETCHER-EXECUTOR[30, 30] [ERROR] INTERRUPTED EXCEPTION CAUGHT IN EXECUTE METHOD 2022-05-30 06:37:07.077 C.D.S.B.FETCHERBOLT THREAD-21-FETCHER-EXECUTOR[30, 30] [ERROR] INTERRUPTED EXCEPTION CAUGHT IN EXECUTE METHOD 2022-05-30 06:37:07.077 C.D.S.B.FETCHERBOLT THREAD-21-FETCHER-EXECUTOR[30, 30] [ERROR] INTERRUPTED EXCEPTION CAUGHT IN EXECUTE METHOD 2022-05-30 06:37:07.077 C.D.S.B.FETCHERBOLT THREAD-21-FETCHER-EXECUTOR[30, 30] [ERROR] INTERRUPTED EXCEPTION CAUGHT IN EXECUTE METHOD

HttpClient.HttpProtocol

We had tried to change the protocol to fix this issue. The CPU has never reach again 100%. But periodically, some fetcher threads are not released.

After some days, those “zombie” threads increase. We are often redeploying the topology (for functional update) and obviously, a new deployment reset thread count.

For now, the issue is less critical then the OkClient one, but we are trying to understand. Do you have any ideas or similar case?

From a user

Links that were once pages and then turn to redirects are our issue. Our content management system auto creates clean URLs. If the title of the page is changed the clean URL is changed and the old URL is redirected to the new URL. The old URL stays in our index unless manually removed. When a link is changed from FETCHED to REDIRECT it would be ideal if the index is removed.

Hi @jnioche,

I was looking into https://github.com/DigitalPebble/storm-crawler/pull/989#discussion_r918581042 and reviewed the old code in order to make sure, that I get the wanted behaviour. (see https://github.com/FelixEngl/storm-crawler/blob/834347e53f79376d3a79f125a6203c91d062e04f/external/elasticsearch/src/main/java/com/digitalpebble/stormcrawler/elasticsearch/bolt/IndexerBolt.java)

Now I am wondering, shouldn't it be enough to only process the first encounter of a BulkResponseElement with a specific id and otherwise just print the required LOG-events and update the counters accordingly?

Because the old code worked like this (if I got that right):

:START afterBulk

:ITERATION 1
+ waitAck ---------------+
| "A" | [tuple1, tuple3] |
| "B" | [tuple2]         |
+------------------------+

+ bulk_response ---------------+
| 1. (id: "A", state: SUCCESS) |
| 2. (id: "B", state: SUCCESS) |
| 3. (id: "A", state: FAILURE) |
+------------------------------+

respone = bulk_respose.removeFirst() : (id: "A", state: SUCCESS)
tuples = waitAck.getIfPresent(response.id) : [tuple1, tuple3]
for(tuple in tuples){
    // process all tuples as state: SUCCESS
    ...
}
waitAck.invalidate(response.id) // Immediate removal
:ITERATION 1

:ITERATION 2
+ waitAck -------+
| "B" | [tuple2] |
+----------------+

+ bulk_response ---------------+
| 2. (id: "B", state: SUCCESS) |
| 3. (id: "A", state: FAILURE) |
+------------------------------+

respone = bulk_respose.removeFirst() : (id: "B", state: SUCCESS)
tuples = waitAck.getIfPresent(response.id) : [tuple2]
for(tuple in tuples){
    // process all tuples as state: SUCCESS
    ...
}
waitAck.invalidate(response.id) // Immediate removal
:ITERATION 2

:ITERATION 3
+ waitAck -------+
+----------------+

+ bulk_response ---------------+
| 3. (id: "A", state: FAILURE) |
+------------------------------+

respone = bulk_respose.removeFirst() : (id: "A", state: FAILURE)
tuples = waitAck.getIfPresent(response.id) : null
LOG.warn("could not find unacked tuple for A")
:ITERATION 3

:STOP afterBulk

Best Regards

Felix

022-07-15 09:57:16.851 o.a.s.e.e.ReportError Thread-43-fetcher-executor[15, 15] [ERROR] Error
java.lang.RuntimeException: java.lang.RuntimeException: java.util.ConcurrentModificationException
	at org.apache.storm.utils.Utils$1.run(Utils.java:411) ~[storm-client-2.4.0.jar:2.4.0]
	at java.lang.Thread.run(Thread.java:829) [?:?]
Caused by: java.lang.RuntimeException: java.util.ConcurrentModificationException
	at org.apache.storm.executor.Executor.accept(Executor.java:301) ~[storm-client-2.4.0.jar:2.4.0]
	at org.apache.storm.utils.JCQueue.consumeImpl(JCQueue.java:113) ~[storm-client-2.4.0.jar:2.4.0]
	at org.apache.storm.utils.JCQueue.consume(JCQueue.java:89) ~[storm-client-2.4.0.jar:2.4.0]
	at org.apache.storm.executor.bolt.BoltExecutor$1.call(BoltExecutor.java:154) ~[storm-client-2.4.0.jar:2.4.0]
	at org.apache.storm.executor.bolt.BoltExecutor$1.call(BoltExecutor.java:140) ~[storm-client-2.4.0.jar:2.4.0]
	at org.apache.storm.utils.Utils$1.run(Utils.java:396) ~[storm-client-2.4.0.jar:2.4.0]
	... 1 more
Caused by: java.util.ConcurrentModificationException
	at java.util.HashMap$HashIterator.nextNode(HashMap.java:1511) ~[?:?]
	at java.util.HashMap$EntryIterator.next(HashMap.java:1544) ~[?:?]
	at java.util.HashMap$EntryIterator.next(HashMap.java:1542) ~[?:?]
	at org.apache.storm.metric.api.MultiCountMetric.getValueAndReset(MultiCountMetric.java:35) ~[storm-client-2.4.0.jar:2.4.0]
	at org.apache.storm.metric.api.MultiCountMetric.getValueAndReset(MultiCountMetric.java:18) ~[storm-client-2.4.0.jar:2.4.0]
	at org.apache.storm.executor.Executor.metricsTick(Executor.java:339) ~[storm-client-2.4.0.jar:2.4.0]
	at org.apache.storm.executor.bolt.BoltExecutor.tupleActionFn(BoltExecutor.java:200) ~[storm-client-2.4.0.jar:2.4.0]
	at org.apache.storm.executor.Executor.accept(Executor.java:297) ~[storm-client-2.4.0.jar:2.4.0]
	at org.apache.storm.utils.JCQueue.consumeImpl(JCQueue.java:113) ~[storm-client-2.4.0.jar:2.4.0]
	at org.apache.storm.utils.JCQueue.consume(JCQueue.java:89) ~[storm-client-2.4.0.jar:2.4.0]
	at org.apache.storm.executor.bolt.BoltExecutor$1.call(BoltExecutor.java:154) ~[storm-client-2.4.0.jar:2.4.0]
	at org.apache.storm.executor.bolt.BoltExecutor$1.call(BoltExecutor.java:140) ~[storm-client-2.4.0.jar:2.4.0]
	at org.apache.storm.utils.Utils$1.run(Utils.java:396) ~[storm-client-2.4.0.jar:2.4.0]
	... 1 more