Merge pull request #72 from agrare/rbvmomi_inventory_collector

Rbvmomi inventory collector

app/models/manageiq/providers/vmware/infra_manager.rb

@@ -9,6 +9,7 @@ class Vmware::InfraManager < InfraManager
  require_nested :Refresher
  require_nested :Host
  require_nested :HostEsx
+ require_nested :Inventory
  require_nested :Provision
  require_nested :ProvisionViaPxe
  require_nested :ProvisionWorkflow

app/models/manageiq/providers/vmware/infra_manager/inventory.rb

@@ -1,3 +1,4 @@
 class ManageIQ::Providers::Vmware::InfraManager::Inventory < ManagerRefresh::Inventory
+ require_nested :Collector
  require_nested :Persister
 end

app/models/manageiq/providers/vmware/infra_manager/inventory/collector.rb

@@ -0,0 +1,175 @@
+class ManageIQ::Providers::Vmware::InfraManager::Inventory::Collector
+ include InventoryCache
+ include PropertyCollector
+ include Vmdb::Logging
+
+ attr_reader :ems, :exit_requested
+ private :ems, :exit_requested
+
+ def initialize(ems)
+ @ems = ems
+ @exit_requested = false
+ end
+
+ def run
+ until exit_requested
+ vim = connect(ems.address, ems.authentication_userid, ems.authentication_password)
+
+ begin
+ wait_for_updates(vim)
+ rescue RbVmomi::Fault => err
+ _log.err("Caught exception #{err.message}")
+ _log.log_backtrace(err)
+ ensure
+ vim.serviceContent.sessionManager.Logout
+ vim = nil
+ end
+ end
+
+ _log.info("Exiting...")
+ ensure
+ vim.serviceContent.sessionManager.Logout unless vim.nil?
+ end
+
+ def stop
+ _log.info("Exit request received...")
+ @exit_requested = true
+ end
+
+ private
+
+ def connect(host, username, password)
+ _log.info("Connecting to #{username}@#{host}...")
+
+ vim_opts = {
+ :ns => 'urn:vim25',
+ :host => host,
+ :ssl => true,
+ :insecure => true,
+ :path => '/sdk',
+ :port => 443,
+ :user => username,
+ :password => password,
+ }
+
+ require 'rbvmomi/vim'
+
+ vim = RbVmomi::VIM.connect(vim_opts)
+
+ _log.info("Connected")
+ vim
+ end
+
+ def wait_for_updates(vim)
+ property_filter = create_property_filter(vim)
+
+ # Return if we don't receive any updates for 60 seconds break
+ # so that we can check if we are supposed to exit
+ options = RbVmomi::VIM.WaitOptions(:maxWaitSeconds => 60)
+
+ # Send the "special initial data version" i.e. an empty string
+ # so that we get all inventory back in the first update set
+ version = ""
+
+ _log.info("Refreshing initial inventory...")
+
+ initial = true
+ until exit_requested
+ update_set = vim.propertyCollector.WaitForUpdatesEx(:version => version, :options => options)
+ next if update_set.nil?
+
+ # Save the new update set version
+ version = update_set.version
+
+ property_filter_update_set = update_set.filterSet
+ next if property_filter_update_set.blank?
+
+ property_filter_update_set.each do |property_filter_update|
+ next if property_filter_update.filter != property_filter
+
+ object_update_set = property_filter_update.objectSet
+ next if object_update_set.blank?
+
+ process_object_update_set(object_update_set)
+ end
+
+ next if update_set.truncated
+
+ next unless initial
+
+ _log.info("Refreshing initial inventory...Complete")
+ initial = false
+ end
+ ensure
+ property_filter.DestroyPropertyFilter unless property_filter.nil?
+ end
+
+ def process_object_update_set(object_update_set)
+ _log.info("Processing #{object_update_set.count} updates...")
+
+ object_update_set.each do |object_update|
+ process_object_update(object_update)
+ end
+
+ _log.info("Processing #{object_update_set.count} updates...Complete")
+ end
+
+ def process_object_update(object_update)
+ managed_object = object_update.obj
+
+ case object_update.kind
+ when "enter", "modify"
+ process_object_update_modify(managed_object, object_update.changeSet)
+ when "leave"
+ process_object_update_leave(managed_object)
+ end
+ end
+
+ def process_object_update_modify(obj, change_set, _missing_set = [])
+ obj_type = obj.class.wsdl_name
+ obj_ref = obj._ref
+
+ props = inventory_cache[obj_type][obj_ref].dup
+
+ change_set.each do |property_change|
+ next if property_change.nil?
+
+ case property_change.op
+ when 'add'
+ process_property_change_add(props, property_change)
+ when 'remove', 'indirectRemove'
+ process_property_change_remove(props, property_change)
+ when 'assign'
+ process_property_change_assign(props, property_change)
+ end
+ end
+
+ update_inventory_cache(obj_type, obj_ref, props)
+
+ props
+ end
+
+ def process_object_update_leave(obj)
+ obj_type = obj.class.wsdl_name
+ obj_ref = obj._ref
+
+ inventory_cache[obj_type].delete(obj_ref)
+
+ nil
+ end
+
+ def process_property_change_add(props, property_change)
+ name = property_change.name
+
+ props[name] ||= []
+ props[name] << property_change.val
+ end
+
+ def process_property_change_remove(props, property_change)
+ props.delete(property_change.name)
+ end
+
+ def process_property_change_assign(props, property_change)
+ props[property_change.name] = property_change.val
+ end
+end

app/models/manageiq/providers/vmware/infra_manager/inventory/collector/inventory_cache.rb

@@ -0,0 +1,39 @@
+class ManageIQ::Providers::Vmware::InfraManager::Inventory::Collector
+ module InventoryCache
+ private
+
+ def inventory_cache
+ @inventory_cache ||=
+ Hash.new do |h, k|
+ h[k] = Hash.new { |h1, k1| h1[k1] = Hash.new }
+ end
+ end
+
+ INVENTORY_CACHE_PROPERTIES = {
+ "VirtualMachine" => %w(
+ summary.config.template
+ summary.config.name
+ summary.config.uuid
+ summary.config.vmPathName
+ ),
+ "Host" => %w(
+ config.network.dnsConfig.hostName
+ summary.config.product.name
+ ),
+ "Datastore" => %w(
+ summary.name
+ summary.url
+ )
+ }.freeze
+
+ def update_inventory_cache(obj_type, obj_ref, props)
+ properties_to_cache = INVENTORY_CACHE_PROPERTIES[obj_type]
+ return if properties_to_cache.blank?
+
+ cache = inventory_cache[obj_type][obj_ref]
+ properties_to_cache.each do |prop_key|
+ cache[prop_key] = props[prop_key]
+ end
+ end
+ end
+end