我们从Python开源项目中，提取了以下19个代码示例，用于说明如何使用libvirt.openReadOnly()。
def __enter__(self): try: if self.sasl_username and self.sasl_password: def request_cred(credentials, user_data): for credential in credentials: if credential[0] == libvirt.VIR_CRED_AUTHNAME: credential[4] = self.sasl_username elif credential[0] == libvirt.VIR_CRED_PASSPHRASE: credential[4] = self.sasl_password return 0 auth = [[libvirt.VIR_CRED_AUTHNAME, libvirt.VIR_CRED_PASSPHRASE], request_cred, None] flags = libvirt.VIR_CONNECT_RO if self.readonly else 0 self.conn = libvirt.openAuth(self.uri, auth, flags) elif self.readonly: self.conn = libvirt.openReadOnly(self.uri) else: self.conn = libvirt.open(self.uri) return self.conn except libvirt.libvirtError as e: raise exception.LibvirtConnectionOpenError(uri=self.uri, error=e)
def get_domain_memory(overhead): """Get memory allocated by domains in MiB -> dict arguments: overhead - MiB qemu overhead per domain """ result = dict() try: con = openReadOnly(None) domains = con.listAllDomains() for domain in domains: name = domain.name() result[name] = float(domain.maxMemory()) / 1024.0 + overhead finally: con.close() return result
def main(): try: conn = openReadOnly(None) except libvirtError as error: print('WARNING: could not connect to libvirt: ' + str(error)) exit(1) inactive_domains = [d for d in conn.listAllDomains() if not d.isActive()] if inactive_domains: print('WARNING: ' + ', '.join( '{} is defined but not running'.format(d.name()) for d in inactive_domains )) exit(1) print('OK: all defined domains are running') exit(0)
def add_hyp_to_receive_events(self, hyp_id): d_hyp_parameters = get_hyp_hostname_user_port_from_id(hyp_id) hostname = d_hyp_parameters['hostname'] user = d_hyp_parameters.get('user', 'root') port = d_hyp_parameters.get('port', 22) uri = hostname_to_uri(hostname, user=user, port=port) conn_ok = False try: self.hyps_conn[hyp_id] = libvirt.openReadOnly(uri) log.debug('####################connection to {} ready in events thread'.format(hyp_id)) update_uri_hyp(hyp_id, uri) conn_ok = True except Exception as e: log.error('libvirt connection read only in events thread in hypervisor: {}'.format(hyp_id)) log.error(e) if conn_ok is True: self.events_ids[hyp_id] = self.register_events(self.hyps_conn[hyp_id]) self.hyps[hyp_id] = hostname
def libvirt_conn(f): @functools.wraps(f) def wrapper(*args, **kwargs): conn = libvirt.openReadOnly('qemu:///system') return f(conn, *args, **kwargs) return wrapper
def __enter__(self): try: self._conn = (libvirt.openReadOnly(self.uri) if self.readonly else libvirt.open(self.uri)) return self._conn except libvirt.libvirtError as e: print('Error when connecting to the libvirt URI "%(uri)s": ' '%(error)s' % {'uri': self.uri, 'error': e}) flask.abort(500)
def kvm_connect(): try: conn = libvirt.openReadOnly('qemu:///system') except: sys.stderr.write( "There was an error connecting to the local libvirt daemon using '" + uri + "'.") exit(1) return conn
def connect(self): """Connect to hypervisor and return object.""" try: if self.read_only: conn = libvirt.openReadOnly(self.uri) else: conn = libvirt.open(self.uri) except: raise return conn
def guest(): guest = {'hosts': [], 'vars': {'ansible_python_interpreter': '/usr/bin/python', 'hv_node_netmask': '255.255.0.0', 'hv_node_broadcast': '10.0.255.255'}} c = libvirt.openReadOnly("qemu:///system") if c != None: for i in c.listDomainsID(): dom = c.lookupByID(i) if dom.name().startswith('hv'): guest['hosts'].append(dom.name()) return guest
def __enter__(self): if not self._conn: self._conn = libvirt.openReadOnly(self._con_str) return self._conn
def open(self): uri = None if self._host != 'localhost': uri = 'qemu+ssh://%s/system' % self._host self.connection = libvirt.open(uri) self.readonly = libvirt.openReadOnly(uri)
def open(self): uri = None self.authorized = False j.tools.prefab.local.system.ssh.keygen(name='libvirt') self.pubkey = j.tools.prefab.local.core.file_read('/root/.ssh/libvirt.pub') if self._host != 'localhost': self.authorized = not self.executor.prefab.system.ssh.authorize(self.user, self.pubkey) uri = 'qemu+ssh://%s/system?no_tty=1&keyfile=/root/.ssh/libvirt&no_verify=1' % self._host self.connection = libvirt.open(uri) self.readonly = libvirt.openReadOnly(uri)
def cli(ctx): """Get metrics from a KVM hypervisor.""" setattr(cli, '__doc__', DOC) # Lower level import because we only want to load this module # when this plugin is called. try: import libvirt except ImportError: raise SystemExit('The "kvm plugin requires libvirt-python to be' ' installed".') output = { 'measurement_name': 'kvm', 'meta': { 'kvm_host_id': abs(hash(socket.getfqdn())) } } # Open a read-only connection to libvirt conn = libvirt.openReadOnly("qemu:///system") try: variables = dict() # Get all of the KVM instances on this host. domains = conn.listDomainsID() variables['kvm_vms'] = len(domains) variables['kvm_total_vcpus'] = conn.getCPUMap()[0] variables['kvm_scheduled_vcpus'] = 0 # Loop through each instance to gather additional data. for domain in domains: variables['kvm_scheduled_vcpus'] += conn.lookupByID( domain ).maxVcpus() # Return the data. output['variables'] = variables except Exception as exp: # We may have picked up an exception while querying libvirt for data. output['exit_code'] = 1 output['message'] = '{} failed -- {}'.format( COMMAND_NAME, utils.log_exception(exp=exp) ) else: output['exit_code'] = 0 output['message'] = 'kvm is ok' finally: conn.close() return output
def _virt_event(self, uri): # Run a background thread with the event loop self._vir_event_loop_native_start() event_callback_handlers = { libvirt.VIR_DOMAIN_EVENT_ID_LIFECYCLE: self._my_domain_event_callback, libvirt.VIR_DOMAIN_EVENT_ID_REBOOT: self._my_domain_event_reboot_callback, libvirt.VIR_DOMAIN_EVENT_ID_RTC_CHANGE: self._my_domain_event_rtc_change_callback, libvirt.VIR_DOMAIN_EVENT_ID_IO_ERROR: self._my_domain_event_io_error_callback, libvirt.VIR_DOMAIN_EVENT_ID_WATCHDOG: self._my_domain_event_watchdog_callback, libvirt.VIR_DOMAIN_EVENT_ID_GRAPHICS: self._my_domain_event_graphics_callback, libvirt.VIR_DOMAIN_EVENT_ID_DISK_CHANGE: self._my_domain_event_disk_change_callback, libvirt.VIR_DOMAIN_EVENT_ID_IO_ERROR_REASON: self._my_domain_event_io_error_reason_callback, libvirt.VIR_DOMAIN_EVENT_ID_CONTROL_ERROR: self._my_domain_event_generic_callback } # Connect to libvirt - If be disconnected, reprocess. self.running = True while self.running: vc = libvirt.openReadOnly(uri) # Event callback settings callback_ids = [] for event, callback in event_callback_handlers.items(): cid = vc.domainEventRegisterAny(None, event, callback, None) callback_ids.append(cid) # Connection monitoring. vc.setKeepAlive(5, 3) while vc.isAlive() == 1 and self.running: eventlet.greenthread.sleep(1) # If connection between libvirtd was lost, # clear callback connection. LOG.warning("Libvirt Connection Closed Unexpectedly.") for cid in callback_ids: try: vc.domainEventDeregisterAny(cid) except Exception: pass vc.close() del vc time.sleep(3)
def get_inventory(self): ''' Construct the inventory ''' inventory = dict(_meta=dict(hostvars=dict())) conn = libvirt.openReadOnly(self.libvirt_uri) if conn is None: print "Failed to open connection to %s" % self.libvirt_uri sys.exit(1) domains = conn.listAllDomains() if domains is None: print "Failed to list domains for connection %s" % self.libvirt_uri sys.exit(1) for domain in domains: hostvars = dict(libvirt_name=domain.name(), libvirt_id=domain.ID(), libvirt_uuid=domain.UUIDString()) domain_name = domain.name() # TODO: add support for guests that are not in a running state state, _ = domain.state() # 2 is the state for a running guest if state != 1: continue hostvars['libvirt_status'] = 'running' root = ET.fromstring(domain.XMLDesc()) ansible_ns = {'ansible': 'https://github.com/ansible/ansible'} for tag_elem in root.findall('./metadata/ansible:tags/ansible:tag', ansible_ns): tag = tag_elem.text _push(inventory, "tag_%s" % tag, domain_name) _push(hostvars, 'libvirt_tags', tag) # TODO: support more than one network interface, also support # interface types other than 'network' interface = root.find("./devices/interface[@type='network']") if interface is not None: source_elem = interface.find('source') mac_elem = interface.find('mac') if source_elem is not None and \ mac_elem is not None: # Adding this to disable pylint check specifically # ignoring libvirt-python versions that # do not include DHCPLeases # This is needed until we upgrade the build bot to # RHEL7 (>= 1.2.6 libvirt) # pylint: disable=no-member dhcp_leases = conn.networkLookupByName(source_elem.get('network')) \ .DHCPLeases(mac_elem.get('address')) if len(dhcp_leases) > 0: ip_address = dhcp_leases[0]['ipaddr'] hostvars['ansible_ssh_host'] = ip_address hostvars['libvirt_ip_address'] = ip_address inventory['_meta']['hostvars'][domain_name] = hostvars return inventory
