← Back to branch summary

~ubuntu-branches/ubuntu/lucid/landscape-client/lucid-updates

« back to all changes in this revision

Viewing changes to landscape/package/reporter.py

  • Committer: Package Import Robot
  • Author(s): Andreas Hasenack
  • Date: 2012-04-10 14:28:48 UTC
  • mfrom: (1.1.27)
  • mto: This revision was merged to the branch mainline in revision 35.
  • Revision ID: package-import@ubuntu.com-20120410142848-7xsy4g2xii7y7ntc
Import upstream version 12.04.3

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/package/reporter.py
10
10
from landscape.lib.twisted_util import gather_results, spawn_process
11
11
from landscape.lib.fetch import fetch_async
12
12
from landscape.lib.fs import touch_file
 
13
from landscape.lib import bpickle
13
14
 
 
15
from landscape.package.facade import AptFacade
14
16
from landscape.package.taskhandler import (
15
17
    PackageTaskHandlerConfiguration, PackageTaskHandler, run_task_handler)
16
 
from landscape.package.store import UnknownHashIDRequest
 
18
from landscape.package.store import UnknownHashIDRequest, FakePackageStore
17
19
 
18
20
 
19
21
HASH_ID_REQUEST_TIMEOUT = 7200
48
50
 
49
51
    smart_update_interval = 60
50
52
    smart_update_filename = "/usr/lib/landscape/smart-update"
 
53
    apt_update_filename = "/usr/lib/landscape/apt-update"
51
54
    sources_list_filename = "/etc/apt/sources.list"
52
55
    sources_list_directory = "/etc/apt/sources.list.d"
53
56
 
54
57
    def run(self):
55
58
        result = Deferred()
56
59
 
57
 
        # Run smart-update before anything else, to make sure that
58
 
        # the SmartFacade will load freshly updated channels
59
 
        result.addCallback(lambda x: self.run_smart_update())
 
60
        if isinstance(self._facade, AptFacade):
 
61
            # Update APT cache if APT facade is enabled.
 
62
            result.addCallback(lambda x: self.run_apt_update())
 
63
        else:
 
64
            # Run smart-update before anything else, to make sure that
 
65
            # the SmartFacade will load freshly updated channels
 
66
            result.addCallback(lambda x: self.run_smart_update())
60
67
 
61
68
        # If the appropriate hash=>id db is not there, fetch it
62
69
        result.addCallback(lambda x: self.fetch_hash_id_db())
79
86
        result.callback(None)
80
87
        return result
81
88
 
 
89
    def send_message(self, message):
 
90
        return self._broker.send_message(message, True)
 
91
 
82
92
    def fetch_hash_id_db(self):
83
93
        """
84
94
        Fetch the appropriate pre-canned database of hash=>id mappings
198
208
                    self.smart_update_filename, code, err))
199
209
            logging.debug("'%s' exited with status %d (out='%s', err='%s'" % (
200
210
                self.smart_update_filename, code, out, err))
201
 
            touch_file(self._config.smart_update_stamp_filename)
 
211
            touch_file(self._config.update_stamp_filename)
 
212
            if not smart_failed and not self._facade.get_channels():
 
213
                code = 1
 
214
                err = "There are no APT sources configured in %s or %s." % (
 
215
                    self.sources_list_filename, self.sources_list_directory)
202
216
            deferred = self._broker.call_if_accepted(
203
217
                "package-reporter-result", self.send_result, code, err)
204
218
            deferred.addCallback(lambda ignore: (out, err, code))
207
221
        result.addCallback(callback)
208
222
        return result
209
223
 
 
224
    def _apt_update_timeout_expired(self, interval):
 
225
        """Check if the apt-update timeout has passed."""
 
226
        stamp = self._config.update_stamp_filename
 
227
 
 
228
        if not os.path.exists(stamp):
 
229
            return True
 
230
        # check stamp file mtime
 
231
        last_update = os.stat(stamp)[8]
 
232
        now = int(time.time())
 
233
        return (last_update + interval * 60) < now
 
234
 
 
235
    def run_apt_update(self):
 
236
        """Run apt-update and log a warning in case of non-zero exit code.
 
237
 
 
238
        @return: a deferred returning (out, err, code)
 
239
        """
 
240
        if (self._config.force_smart_update or self._apt_sources_have_changed()
 
241
            or self._apt_update_timeout_expired(self.smart_update_interval)):
 
242
 
 
243
            result = spawn_process(self.apt_update_filename)
 
244
 
 
245
            def callback((out, err, code)):
 
246
                touch_file(self._config.update_stamp_filename)
 
247
                logging.debug(
 
248
                    "'%s' exited with status %d (out='%s', err='%s')" % (
 
249
                        self.apt_update_filename, code, out, err))
 
250
 
 
251
                if code != 0:
 
252
                    logging.warning("'%s' exited with status %d (%s)" % (
 
253
                            self.apt_update_filename, code, err))
 
254
                elif not self._facade.get_channels():
 
255
                    code = 1
 
256
                    err = ("There are no APT sources configured in %s or %s." %
 
257
                           (self.sources_list_filename,
 
258
                            self.sources_list_directory))
 
259
 
 
260
                deferred = self._broker.call_if_accepted(
 
261
                    "package-reporter-result", self.send_result, code, err)
 
262
                deferred.addCallback(lambda ignore: (out, err, code))
 
263
                return deferred
 
264
 
 
265
            return result.addCallback(callback)
 
266
 
 
267
        else:
 
268
            logging.debug("'%s' didn't run, update interval has not passed" %
 
269
                          self.apt_update_filename)
 
270
            return succeed(("", "", 0))
 
271
 
210
272
    def send_result(self, code, err):
211
273
        """
212
274
        Report the package reporter result to the server in a message.
215
277
            "type": "package-reporter-result",
216
278
            "code": code,
217
279
            "err": err}
218
 
        return self._broker.send_message(message, True)
 
280
        return self.send_message(message)
219
281
 
220
282
    def handle_task(self, task):
221
283
        message = task.data
385
447
        """Create a hash_id_request and send message with "request-id"."""
386
448
        request = self._store.add_hash_id_request(unknown_hashes)
387
449
        message["request-id"] = request.id
388
 
        result = self._broker.send_message(message, True)
 
450
        result = self.send_message(message)
389
451
 
390
452
        def set_message_id(message_id):
391
453
            request.message_id = message_id
454
516
            hash = self._facade.get_package_hash(package)
455
517
            id = self._store.get_hash_id(hash)
456
518
            if id is not None:
457
 
                if package.installed:
 
519
                if self._facade.is_package_installed(package):
458
520
                    current_installed.add(id)
459
 
                    for loader in package.loaders:
460
 
                        # Is the package also in a non-installed
461
 
                        # loader?  IOW, "available".
462
 
                        if not loader.getInstalled():
463
 
                            current_available.add(id)
464
 
                            break
 
521
                    if self._facade.is_package_available(package):
 
522
                        current_available.add(id)
465
523
                else:
466
524
                    current_available.add(id)
467
525
 
468
526
                # Are there any packages that this package is an upgrade for?
469
 
                for upgrade in package.upgrades:
470
 
                    for provides in upgrade.providedby:
471
 
                        for provides_package in provides.packages:
472
 
                            if provides_package.installed:
473
 
                                current_upgrades.add(id)
474
 
                                break
475
 
                        else:
476
 
                            continue
477
 
                        break
478
 
                    else:
479
 
                        continue
480
 
                    break
 
527
                if self._facade.is_package_upgrade(package):
 
528
                    current_upgrades.add(id)
481
529
 
482
530
        for package in self._facade.get_locked_packages():
483
531
            hash = self._facade.get_package_hash(package)
526
574
            return succeed(False)
527
575
 
528
576
        message["type"] = "packages"
529
 
        result = self._broker.send_message(message, True)
 
577
        result = self.send_message(message)
530
578
 
531
579
        logging.info("Queuing message with changes in known packages: "
532
580
                     "%d installed, %d available, %d available upgrades, "
591
639
            return succeed(False)
592
640
 
593
641
        message["type"] = "package-locks"
594
 
        result = self._broker.send_message(message, True)
 
642
        result = self.send_message(message)
595
643
 
596
644
        logging.info("Queuing message with changes in known package locks:"
597
645
                     " %d created, %d deleted." %
609
657
        return result
610
658
 
611
659
 
 
660
class FakeGlobalReporter(PackageReporter):
 
661
    """
 
662
    A standard reporter, which additionally stores messages sent into its
 
663
    package store.
 
664
    """
 
665
 
 
666
    package_store_class = FakePackageStore
 
667
 
 
668
    def send_message(self, message):
 
669
        self._store.save_message(message)
 
670
        return super(FakeGlobalReporter, self).send_message(message)
 
671
 
 
672
 
 
673
class FakeReporter(PackageReporter):
 
674
    """
 
675
    A fake reporter which only sends messages previously stored by a
 
676
    L{FakeGlobalReporter}.
 
677
    """
 
678
 
 
679
    package_store_class = FakePackageStore
 
680
    global_store_filename = None
 
681
 
 
682
    def run(self):
 
683
        result = succeed(None)
 
684
 
 
685
        # If the appropriate hash=>id db is not there, fetch it
 
686
        result.addCallback(lambda x: self.fetch_hash_id_db())
 
687
 
 
688
        result.addCallback(lambda x: self._store.clear_tasks())
 
689
 
 
690
        # Finally, verify if we have anything new to send to the server.
 
691
        result.addCallback(lambda x: self.send_pending_messages())
 
692
 
 
693
        return result
 
694
 
 
695
    def send_pending_messages(self):
 
696
        """
 
697
        As the last callback of L{PackageReporter}, sends messages stored.
 
698
        """
 
699
        if self.global_store_filename is None:
 
700
            self.global_store_filename = os.environ["FAKE_PACKAGE_STORE"]
 
701
        if not os.path.exists(self.global_store_filename):
 
702
            return succeed(None)
 
703
        message_sent = set(self._store.get_message_ids())
 
704
        global_store = FakePackageStore(self.global_store_filename)
 
705
        all_message_ids = set(global_store.get_message_ids())
 
706
        not_sent = all_message_ids - message_sent
 
707
        deferred = succeed(None)
 
708
        got_type = set()
 
709
        if not_sent:
 
710
            messages = global_store.get_messages_by_ids(not_sent)
 
711
            sent = []
 
712
            for message_id, message in messages:
 
713
                message = bpickle.loads(str(message))
 
714
                if message["type"] not in got_type:
 
715
                    got_type.add(message["type"])
 
716
                    sent.append(message_id)
 
717
                    deferred.addCallback(
 
718
                        lambda x, message=message: self.send_message(message))
 
719
            self._store.save_message_ids(sent)
 
720
        return deferred
 
721
 
 
722
 
612
723
def main(args):
613
 
    return run_task_handler(PackageReporter, args)
 
724
    if "FAKE_GLOBAL_PACKAGE_STORE" in os.environ:
 
725
        return run_task_handler(FakeGlobalReporter, args)
 
726
    elif "FAKE_PACKAGE_STORE" in os.environ:
 
727
        return run_task_handler(FakeReporter, args)
 
728
    else:
 
729
        return run_task_handler(PackageReporter, args)
614
730
 
615
731
 
616
732
def find_reporter_command():