1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
"""Implementation of the "poller" IPC component"""
import argparse
import os
import sys
import time
import systemd.daemon
import gitubuntu.importer_service as importer_service
import gitubuntu.scriptutils as scriptutils
from gitubuntu.source_information import launchpad_login
def main_loop(
lp,
service_state,
poll_interval,
filter_func=None,
):
"""Main loop for the importer service poller process
:param importer_service.State service_state: the state object of the
importer service.
:param int poll_interval: how often to poll, in seconds.
:param filter_func: a function that when called with a sole parameter
string providing a package name, returns a bool that specifies whether
or not that package should be imported.
"""
while True:
# Notify systemd that we're still alive
systemd.daemon.notify('WATCHDOG=1')
importer_service.request_new_imports(
lp=lp,
num_days_ago=1,
service_state=service_state,
filter_func=filter_func,
)
time.sleep(poll_interval)
def main(
data_directory,
denylist_path,
poll_interval,
):
"""Entry point for the importer service poller process
:param str data_directory: the path where persistent data on the status of
the import may be stored
:param str denylist_path: filesystem path to a text file of packages
to never import
:param int poll_interval: how often to poll, in seconds.
"""
service_state = importer_service.State(os.path.join(data_directory, 'db'))
denylist = frozenset(importer_service.read_package_list(denylist_path))
# Initialize system notification
systemd.daemon.notify('READY=1')
print("Monitoring packages", file=sys.stderr)
sys.stderr.flush() # systemd-run services buffer stderr by default
lp = launchpad_login()
main_loop(
lp=lp,
service_state=service_state,
poll_interval=poll_interval,
filter_func=lambda package: package not in denylist,
) # never returns
def parse_args(subparsers=None, base_subparsers=None):
kwargs = {
'description': (
"Monitor Launchpad for new publications and request imports "
"accordingly"
),
'formatter_class': argparse.ArgumentDefaultsHelpFormatter,
}
if base_subparsers:
kwargs['parents'] = base_subparsers
if subparsers:
parser = subparsers.add_parser('importer-service-poller', **kwargs)
parser.set_defaults(func=cli_main)
else:
parser = argparse.ArgumentParser(**kwargs)
parser.add_argument(
'--data-directory',
help="Directory to store persistent data",
default='/var/local/git-ubuntu',
)
parser.add_argument(
'--denylist',
type=str,
help="Path to denylist file",
default=scriptutils.DEFAULTS.denylist,
)
parser.add_argument(
'--poll-interval',
type=int,
help=(
"Polling interval in seconds to examine Launchpad for new "
"publications"
),
default=600,
)
if not subparsers:
return parser.parse_args()
return 'import-service-poller - %s' % kwargs['description']
def cli_main(args):
"""CLI entry point for the importer service poller process"""
main(
data_directory=args.data_directory,
denylist_path=args.denylist,
poll_interval=args.poll_interval,
)
|
