defmodule Crawly.Manager do
@moduledoc """
Crawler manager module
This module is responsible for spawning all processes related to
a given Crawler.
The manager spawns the following processes tree.
┌────────────────┐ ┌───────────────────┐
│ Crawly.Manager ├────────> Crawly.ManagerSup │
└────────────────┘ └─────────┬─────────┘
│ |
│ |
┌──────────────────────────┤
│ │
│ │
┌────────▼───────┐ ┌─────────▼───────┐
│ Worker1 │ │ Worker2 │
└────────┬───────┘ └────────┬────────┘
│ │
│ │
│ │
│ │
┌────────▼─────────┐ ┌──────────▼───────────┐
│Crawly.DataStorage│ │Crawly.RequestsStorage│
└──────────────────┘ └──────────────────────┘
"""
require Logger
@timeout 60_000
@start_request_split_size 50
use GenServer
alias Crawly.{Engine, Utils}
@spec add_workers(module(), non_neg_integer()) ::
:ok | {:error, :spider_non_exist}
def add_workers(spider_name, num_of_workers) do
case Engine.get_manager(spider_name) do
{:error, reason} ->
{:error, reason}
pid ->
GenServer.cast(pid, {:add_workers, num_of_workers})
end
end
def start_link([spider_name, options]) do
Logger.debug("Starting the manager for #{inspect(spider_name)}")
GenServer.start_link(__MODULE__, [spider_name, options])
end
@impl true
def init([spider_name, options]) do
crawl_id = Keyword.get(options, :crawl_id)
Logger.metadata(spider_name: spider_name, crawl_id: crawl_id)
itemcount_limit =
Keyword.get(
options,
:closespider_itemcount,
get_default_limit(:closespider_itemcount, spider_name)
)
closespider_timeout_limit =
Keyword.get(
options,
:closespider_timeout,
get_default_limit(:closespider_timeout, spider_name)
)
# Start DataStorage worker
{:ok, data_storage_pid} =
Crawly.DataStorage.start_worker(spider_name, crawl_id)
Process.link(data_storage_pid)
# Start RequestsWorker for a given spider
{:ok, request_storage_pid} =
Crawly.RequestsStorage.start_worker(spider_name, crawl_id)
Process.link(request_storage_pid)
# Start workers
num_workers =
Keyword.get(
options,
:concurrent_requests_per_domain,
Utils.get_settings(:concurrent_requests_per_domain, spider_name, 4)
)
worker_pids =
Enum.map(1..num_workers, fn _x ->
DynamicSupervisor.start_child(
spider_name,
{Crawly.Worker, [spider_name: spider_name, crawl_id: crawl_id]}
)
end)
# Schedule basic service operations for given spider manager
timeout =
Utils.get_settings(:manager_operations_timeout, spider_name, @timeout)
tref = Process.send_after(self(), :operations, timeout)
Logger.debug(
"Started #{Enum.count(worker_pids)} workers for #{inspect(spider_name)}"
)
{:ok,
%{
name: spider_name,
crawl_id: crawl_id,
itemcount_limit: itemcount_limit,
closespider_timeout_limit: closespider_timeout_limit,
tref: tref,
prev_scraped_cnt: 0,
workers: worker_pids
}, {:continue, {:startup, options}}}
end
@impl true
def handle_continue({:startup, options}, state) do
# Add start requests to the requests storage
init = state.name.init(options)
start_requests_from_req = Keyword.get(init, :start_requests, [])
start_requests_from_urls =
init
|> Keyword.get(:start_urls, [])
|> Crawly.Utils.requests_from_urls()
start_requests = start_requests_from_req ++ start_requests_from_urls
# Split start requests, so it's possible to initialize a part of them in async
# manner
{start_reqs, async_start_reqs} =
Enum.split(start_requests, @start_request_split_size)
:ok = Crawly.RequestsStorage.store(state.name, start_reqs)
Task.start(fn ->
Crawly.RequestsStorage.store(state.name, async_start_reqs)
end)
{:noreply, state}
end
@impl true
def handle_cast({:add_workers, num_of_workers}, state) do
Logger.info("Adding #{num_of_workers} workers for #{inspect(state.name)}")
Enum.each(1..num_of_workers, fn _ ->
DynamicSupervisor.start_child(
state.name,
{Crawly.Worker, [spider_name: state.name, crawl_id: state.crawl_id]}
)
end)
{:noreply, state}
end
@impl true
def handle_info(:operations, state) do
Process.cancel_timer(state.tref)
# Close spider if required items count was reached.
{:stored_items, items_count} = Crawly.DataStorage.stats(state.name)
{:stored_requests, requests_count} =
Crawly.RequestsStorage.stats(state.name)
delta = items_count - state.prev_scraped_cnt
Logger.info("Current crawl speed is: #{delta} items/min")
Logger.info("Current requests count is: #{requests_count}")
stop_spider_by_requests_count_and_delta(
state.name,
requests_count,
delta
)
maybe_stop_spider_by_itemcount_limit(
state.name,
items_count,
state.itemcount_limit
)
# Close spider in case if it's not scraping items fast enough
maybe_stop_spider_by_timeout(
state.name,
delta,
state.closespider_timeout_limit
)
tref =
Process.send_after(
self(),
:operations,
Utils.get_settings(:manager_operations_timeout, state.name, @timeout)
)
{:noreply, %{state | tref: tref, prev_scraped_cnt: items_count}}
end
defp stop_spider_by_requests_count_and_delta(
spider_name,
requests_count,
delta
)
when requests_count == 0 and delta == 0 do
Logger.info("Stopping #{inspect(spider_name)}, all requests handled")
Crawly.Engine.stop_spider(spider_name, :spider_finished)
end
defp stop_spider_by_requests_count_and_delta(_, _, _), do: :ok
defp maybe_stop_spider_by_itemcount_limit(
spider_name,
current,
limit
)
when current >= limit do
Logger.info(
"Stopping #{inspect(spider_name)}, closespider_itemcount achieved"
)
Crawly.Engine.stop_spider(spider_name, :itemcount_limit)
end
defp maybe_stop_spider_by_itemcount_limit(_, _, _), do: :ok
defp maybe_stop_spider_by_timeout(spider_name, current, limit)
when current <= limit and is_integer(limit) do
Logger.info("Stopping #{inspect(spider_name)}, itemcount timeout achieved")
Crawly.Engine.stop_spider(spider_name, :itemcount_timeout)
end
defp maybe_stop_spider_by_timeout(_, _, _), do: :ok
defp maybe_convert_to_integer(value) when is_atom(value), do: value
defp maybe_convert_to_integer(value) when is_binary(value),
do: String.to_integer(value)
defp maybe_convert_to_integer(value) when is_integer(value), do: value
# Get a closespider_itemcount or closespider_timeout_limit from config or spider
# settings.
defp get_default_limit(limit_name, spider_name) do
limit_name
|> Utils.get_settings(spider_name)
|> maybe_convert_to_integer()
end
end
