Page MenuHomeSoftware Heritage
Differential D844

Add an indexer worker.
ClosedPublic
Actions

Authored by vlorentz on Dec 18 2018, 2:49 PM.

Details

Reviewers
ardumont
Group Reviewers
Reviewers
Commits
rDENV1a7f18fe977e: Add an indexer worker.
rCDFD1a7f18fe977e: Add an indexer worker.
Test Plan

not tested yet

Diff Detail

Repository
rCDFD Dockerfiles for developers
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

vlorentz created this revision.Dec 18 2018, 2:49 PM
Herald added a reviewer: Reviewers. · View Herald TranscriptDec 18 2018, 2:49 PM
Harbormaster completed remote builds in B3127: Diff 2695.Dec 18 2018, 2:49 PM
vlorentz updated this revision to Diff 2696.Dec 18 2018, 2:51 PM
  • fix commit
Harbormaster completed remote builds in B3128: Diff 2696.Dec 18 2018, 2:51 PM
vlorentz updated this revision to Diff 2697.Dec 18 2018, 2:52 PM
  • fix commit (again)
Harbormaster completed remote builds in B3129: Diff 2697.Dec 18 2018, 2:52 PM
ardumont accepted this revision.Dec 18 2018, 3:08 PM
ardumont added a subscriber: ardumont.

Sounds good to go.

Just some question though, wasn't this blocked on the scheduler part first?

dockerfiles/swh-indexer-worker/indexer_storage.yml
4 ↗(On Diff #2697)

Where does the service come from?

What is installed prior to this diff (do not remember)?

This revision is now accepted and ready to land.Dec 18 2018, 3:08 PM
vlorentz updated this revision to Diff 2702.Dec 18 2018, 3:10 PM
  • remove useless file
Harbormaster completed remote builds in B3134: Diff 2702.Dec 18 2018, 3:10 PM
vlorentz marked an inline comment as done.Dec 18 2018, 3:10 PM
vlorentz added inline comments.
dockerfiles/swh-indexer-worker/indexer_storage.yml
4 ↗(On Diff #2697)

I didn't mean to commit this file. Removed.

vlorentz updated this revision to Diff 2703.Dec 18 2018, 3:10 PM
  • squash
Harbormaster completed remote builds in B3135: Diff 2703.Dec 18 2018, 3:10 PM
Closed by commit rCDFD1a7f18fe977e: Add an indexer worker. (authored by vlorentz). · Explain WhyDec 18 2018, 3:10 PM
This revision was automatically updated to reflect the committed changes.
vlorentz added a commit: rDENV1a7f18fe977e: Add an indexer worker..Nov 26 2019, 1:34 PM

Revision Contents
Changeset List

PathSize
15 lines
dockerfiles/
swh-indexer-worker/
32 lines
12 lines
3 lines

Diff 2705

View Options

docker-compose.yml

Show First 20 Lines • Show All 321 Lines • ▼ Show 20 Lines ) -> List[ListedOrigin]:
"""Get at most the `count` next origins that need to be visited with """Get at most the `count` next origins that need to be visited with
the `visit_type` loader according to the given scheduling `policy`. the `visit_type` loader according to the given scheduling `policy`.
This will mark the origins as "being visited" in the listed_origins This will mark the origins as "being visited" in the listed_origins
table, to avoid scheduling multiple visits to the same origin. table, to avoid scheduling multiple visits to the same origin.
""" """
origin_select_cols = ", ".join(ListedOrigin.select_columns()) origin_select_cols = ", ".join(ListedOrigin.select_columns())
# TODO: filter on last_scheduled "too recent" to avoid always
# re-scheduling the same tasks.
ardumontUnsubmitted
Not Done
Inline Actions

what does 'enabled' mean here?

I gather that ends up in the query like "WHERE enabled AND visit_type=%s"
but i don't know what that means exactly.

ardumont: what does 'enabled' mean here? I gather that ends up in the query like "WHERE enabled AND…
vlorentzAuthorUnsubmitted
Done
Inline Actions

Whether this origin has been seen during the last listing, and visits should be scheduled

vlorentz: Whether this origin has been seen during the last listing, and visits should be scheduled
olasdUnsubmitted
Not Done
Inline Actions

I don't think the enabled field is ever updated currently. But we will, eventually.

Obviously this would deserve a comment rather than being snuck in.

olasd: I don't think the `enabled` field is ever updated currently. But we will, eventually.
where_clauses = [
"enabled", # "NOT enabled" = the lister said the origin no longer exists
"visit_type = %s",
]
if policy == "oldest_scheduled_first": if policy == "oldest_scheduled_first":
query = f""" order_by = "origin_visit_stats.last_scheduled NULLS FIRST"
WITH selected_origins AS ( else:
raise UnknownPolicy(f"Unknown scheduling policy {policy}")
select_query = f"""
SELECT SELECT
{origin_select_cols} {origin_select_cols}
FROM FROM
listed_origins listed_origins
LEFT JOIN LEFT JOIN
origin_visit_stats USING (url, visit_type) origin_visit_stats USING (url, visit_type)
WHERE WHERE
visit_type = %s {" AND ".join(where_clauses)}
ORDER BY ORDER BY
origin_visit_stats.last_scheduled NULLS FIRST {order_by}
LIMIT %s LIMIT %s
"""
query = f"""
WITH selected_origins AS (
{select_query}
), ),
update_stats AS ( update_stats AS (
INSERT INTO INSERT INTO
origin_visit_stats ( origin_visit_stats (
url, visit_type, last_scheduled url, visit_type, last_scheduled
) )
SELECT SELECT
url, visit_type, now() url, visit_type, now()
FROM FROM
selected_origins selected_origins
ON CONFLICT (url, visit_type) DO UPDATE ON CONFLICT (url, visit_type) DO UPDATE
SET last_scheduled = GREATEST(origin_visit_stats.last_scheduled, EXCLUDED.last_scheduled) SET last_scheduled = GREATEST(
origin_visit_stats.last_scheduled,
EXCLUDED.last_scheduled
)
) )
SELECT SELECT
* *
FROM FROM
selected_origins selected_origins
""" # noqa """
cur.execute(query, (visit_type, count))
cur.execute(query, (visit_type, count))
return [ListedOrigin(**d) for d in cur] return [ListedOrigin(**d) for d in cur]
else:
raise UnknownPolicy(f"Unknown scheduling policy {policy}")
task_create_keys = [ task_create_keys = [
"type", "type",
"arguments", "arguments",
"next_run", "next_run",
"policy", "policy",
"status", "status",
"retries_left", "retries_left",
▲ Show 20 Lines • Show All 567 Lines • Show Last 20 Lines
View Options

dockerfiles/swh-indexer-worker/entrypoint.sh

Show First 20 Lines • Show All 321 Lines • ▼ Show 20 Lines ) -> List[ListedOrigin]:
"""Get at most the `count` next origins that need to be visited with """Get at most the `count` next origins that need to be visited with
the `visit_type` loader according to the given scheduling `policy`. the `visit_type` loader according to the given scheduling `policy`.
This will mark the origins as "being visited" in the listed_origins This will mark the origins as "being visited" in the listed_origins
table, to avoid scheduling multiple visits to the same origin. table, to avoid scheduling multiple visits to the same origin.
""" """
origin_select_cols = ", ".join(ListedOrigin.select_columns()) origin_select_cols = ", ".join(ListedOrigin.select_columns())
# TODO: filter on last_scheduled "too recent" to avoid always
# re-scheduling the same tasks.
ardumontUnsubmitted
Not Done
Inline Actions

what does 'enabled' mean here?

I gather that ends up in the query like "WHERE enabled AND visit_type=%s"
but i don't know what that means exactly.

ardumont: what does 'enabled' mean here? I gather that ends up in the query like "WHERE enabled AND…
vlorentzAuthorUnsubmitted
Done
Inline Actions

Whether this origin has been seen during the last listing, and visits should be scheduled

vlorentz: Whether this origin has been seen during the last listing, and visits should be scheduled
olasdUnsubmitted
Not Done
Inline Actions

I don't think the enabled field is ever updated currently. But we will, eventually.

Obviously this would deserve a comment rather than being snuck in.

olasd: I don't think the `enabled` field is ever updated currently. But we will, eventually.
where_clauses = [
"enabled", # "NOT enabled" = the lister said the origin no longer exists
"visit_type = %s",
]
if policy == "oldest_scheduled_first": if policy == "oldest_scheduled_first":
query = f""" order_by = "origin_visit_stats.last_scheduled NULLS FIRST"
WITH selected_origins AS ( else:
raise UnknownPolicy(f"Unknown scheduling policy {policy}")
select_query = f"""
SELECT SELECT
{origin_select_cols} {origin_select_cols}
FROM FROM
listed_origins listed_origins
LEFT JOIN LEFT JOIN
origin_visit_stats USING (url, visit_type) origin_visit_stats USING (url, visit_type)
WHERE WHERE
visit_type = %s {" AND ".join(where_clauses)}
ORDER BY ORDER BY
origin_visit_stats.last_scheduled NULLS FIRST {order_by}
LIMIT %s LIMIT %s
"""
query = f"""
WITH selected_origins AS (
{select_query}
), ),
update_stats AS ( update_stats AS (
INSERT INTO INSERT INTO
origin_visit_stats ( origin_visit_stats (
url, visit_type, last_scheduled url, visit_type, last_scheduled
) )
SELECT SELECT
url, visit_type, now() url, visit_type, now()
FROM FROM
selected_origins selected_origins
ON CONFLICT (url, visit_type) DO UPDATE ON CONFLICT (url, visit_type) DO UPDATE
SET last_scheduled = GREATEST(origin_visit_stats.last_scheduled, EXCLUDED.last_scheduled) SET last_scheduled = GREATEST(
origin_visit_stats.last_scheduled,
EXCLUDED.last_scheduled
)
) )
SELECT SELECT
* *
FROM FROM
selected_origins selected_origins
""" # noqa """
cur.execute(query, (visit_type, count))
cur.execute(query, (visit_type, count))
return [ListedOrigin(**d) for d in cur] return [ListedOrigin(**d) for d in cur]
else:
raise UnknownPolicy(f"Unknown scheduling policy {policy}")
task_create_keys = [ task_create_keys = [
"type", "type",
"arguments", "arguments",
"next_run", "next_run",
"policy", "policy",
"status", "status",
"retries_left", "retries_left",
▲ Show 20 Lines • Show All 567 Lines • Show Last 20 Lines
View Options

dockerfiles/swh-indexer-worker/indexer.yml

Show First 20 Lines • Show All 321 Lines • ▼ Show 20 Lines ) -> List[ListedOrigin]:
"""Get at most the `count` next origins that need to be visited with """Get at most the `count` next origins that need to be visited with
the `visit_type` loader according to the given scheduling `policy`. the `visit_type` loader according to the given scheduling `policy`.
This will mark the origins as "being visited" in the listed_origins This will mark the origins as "being visited" in the listed_origins
table, to avoid scheduling multiple visits to the same origin. table, to avoid scheduling multiple visits to the same origin.
""" """
origin_select_cols = ", ".join(ListedOrigin.select_columns()) origin_select_cols = ", ".join(ListedOrigin.select_columns())
# TODO: filter on last_scheduled "too recent" to avoid always
# re-scheduling the same tasks.
ardumontUnsubmitted
Not Done
Inline Actions

what does 'enabled' mean here?

I gather that ends up in the query like "WHERE enabled AND visit_type=%s"
but i don't know what that means exactly.

ardumont: what does 'enabled' mean here? I gather that ends up in the query like "WHERE enabled AND…
vlorentzAuthorUnsubmitted
Done
Inline Actions

Whether this origin has been seen during the last listing, and visits should be scheduled

vlorentz: Whether this origin has been seen during the last listing, and visits should be scheduled
olasdUnsubmitted
Not Done
Inline Actions

I don't think the enabled field is ever updated currently. But we will, eventually.

Obviously this would deserve a comment rather than being snuck in.

olasd: I don't think the `enabled` field is ever updated currently. But we will, eventually.
where_clauses = [
"enabled", # "NOT enabled" = the lister said the origin no longer exists
"visit_type = %s",
]
if policy == "oldest_scheduled_first": if policy == "oldest_scheduled_first":
query = f""" order_by = "origin_visit_stats.last_scheduled NULLS FIRST"
WITH selected_origins AS ( else:
raise UnknownPolicy(f"Unknown scheduling policy {policy}")
select_query = f"""
SELECT SELECT
{origin_select_cols} {origin_select_cols}
FROM FROM
listed_origins listed_origins
LEFT JOIN LEFT JOIN
origin_visit_stats USING (url, visit_type) origin_visit_stats USING (url, visit_type)
WHERE WHERE
visit_type = %s {" AND ".join(where_clauses)}
ORDER BY ORDER BY
origin_visit_stats.last_scheduled NULLS FIRST {order_by}
LIMIT %s LIMIT %s
"""
query = f"""
WITH selected_origins AS (
{select_query}
), ),
update_stats AS ( update_stats AS (
INSERT INTO INSERT INTO
origin_visit_stats ( origin_visit_stats (
url, visit_type, last_scheduled url, visit_type, last_scheduled
) )
SELECT SELECT
url, visit_type, now() url, visit_type, now()
FROM FROM
selected_origins selected_origins
ON CONFLICT (url, visit_type) DO UPDATE ON CONFLICT (url, visit_type) DO UPDATE
SET last_scheduled = GREATEST(origin_visit_stats.last_scheduled, EXCLUDED.last_scheduled) SET last_scheduled = GREATEST(
origin_visit_stats.last_scheduled,
EXCLUDED.last_scheduled
)
) )
SELECT SELECT
* *
FROM FROM
selected_origins selected_origins
""" # noqa """
cur.execute(query, (visit_type, count))
cur.execute(query, (visit_type, count))
return [ListedOrigin(**d) for d in cur] return [ListedOrigin(**d) for d in cur]
else:
raise UnknownPolicy(f"Unknown scheduling policy {policy}")
task_create_keys = [ task_create_keys = [
"type", "type",
"arguments", "arguments",
"next_run", "next_run",
"policy", "policy",
"status", "status",
"retries_left", "retries_left",
▲ Show 20 Lines • Show All 567 Lines • Show Last 20 Lines
View Options

indexers.env

Show First 20 Lines • Show All 321 Lines • ▼ Show 20 Lines ) -> List[ListedOrigin]:
"""Get at most the `count` next origins that need to be visited with """Get at most the `count` next origins that need to be visited with
the `visit_type` loader according to the given scheduling `policy`. the `visit_type` loader according to the given scheduling `policy`.
This will mark the origins as "being visited" in the listed_origins This will mark the origins as "being visited" in the listed_origins
table, to avoid scheduling multiple visits to the same origin. table, to avoid scheduling multiple visits to the same origin.
""" """
origin_select_cols = ", ".join(ListedOrigin.select_columns()) origin_select_cols = ", ".join(ListedOrigin.select_columns())
# TODO: filter on last_scheduled "too recent" to avoid always
# re-scheduling the same tasks.
ardumontUnsubmitted
Not Done
Inline Actions

what does 'enabled' mean here?

I gather that ends up in the query like "WHERE enabled AND visit_type=%s"
but i don't know what that means exactly.

ardumont: what does 'enabled' mean here? I gather that ends up in the query like "WHERE enabled AND…
vlorentzAuthorUnsubmitted
Done
Inline Actions

Whether this origin has been seen during the last listing, and visits should be scheduled

vlorentz: Whether this origin has been seen during the last listing, and visits should be scheduled
olasdUnsubmitted
Not Done
Inline Actions

I don't think the enabled field is ever updated currently. But we will, eventually.

Obviously this would deserve a comment rather than being snuck in.

olasd: I don't think the `enabled` field is ever updated currently. But we will, eventually.
where_clauses = [
"enabled", # "NOT enabled" = the lister said the origin no longer exists
"visit_type = %s",
]
if policy == "oldest_scheduled_first": if policy == "oldest_scheduled_first":
query = f""" order_by = "origin_visit_stats.last_scheduled NULLS FIRST"
WITH selected_origins AS ( else:
raise UnknownPolicy(f"Unknown scheduling policy {policy}")
select_query = f"""
SELECT SELECT
{origin_select_cols} {origin_select_cols}
FROM FROM
listed_origins listed_origins
LEFT JOIN LEFT JOIN
origin_visit_stats USING (url, visit_type) origin_visit_stats USING (url, visit_type)
WHERE WHERE
visit_type = %s {" AND ".join(where_clauses)}
ORDER BY ORDER BY
origin_visit_stats.last_scheduled NULLS FIRST {order_by}
LIMIT %s LIMIT %s
"""
query = f"""
WITH selected_origins AS (
{select_query}
), ),
update_stats AS ( update_stats AS (
INSERT INTO INSERT INTO
origin_visit_stats ( origin_visit_stats (
url, visit_type, last_scheduled url, visit_type, last_scheduled
) )
SELECT SELECT
url, visit_type, now() url, visit_type, now()
FROM FROM
selected_origins selected_origins
ON CONFLICT (url, visit_type) DO UPDATE ON CONFLICT (url, visit_type) DO UPDATE
SET last_scheduled = GREATEST(origin_visit_stats.last_scheduled, EXCLUDED.last_scheduled) SET last_scheduled = GREATEST(
origin_visit_stats.last_scheduled,
EXCLUDED.last_scheduled
)
) )
SELECT SELECT
* *
FROM FROM
selected_origins selected_origins
""" # noqa """
cur.execute(query, (visit_type, count))
cur.execute(query, (visit_type, count))
return [ListedOrigin(**d) for d in cur] return [ListedOrigin(**d) for d in cur]
else:
raise UnknownPolicy(f"Unknown scheduling policy {policy}")
task_create_keys = [ task_create_keys = [
"type", "type",
"arguments", "arguments",
"next_run", "next_run",
"policy", "policy",
"status", "status",
"retries_left", "retries_left",
▲ Show 20 Lines • Show All 567 Lines • Show Last 20 Lines
About · Developer information · Legal