Add hypervisors and host aggregates

plugin/plugin.go

@@ -16,8 +16,10 @@ func Plugin() *source.Plugin {
 		"github.com/dihedron-openstack",
 		Version,
 		schema.Tables{
+			resources.Aggregates(),
 			resources.Attachments(),
 			resources.Flavors(),
+			resources.Hypervisors(),
 			resources.Instances(),
 			resources.Images(),
 			resources.Networks(),

resources/aggregate_hosts.go

@@ -0,0 +1,34 @@
+package resources
+
+import (
+	"context"
+
+	"github.com/cloudquery/plugin-sdk/schema"
+	"github.com/cloudquery/plugin-sdk/transformers"
+	"github.com/dihedron/cq-plugin-utils/transform"
+	"github.com/dihedron/cq-source-openstack/client"
+	"github.com/gophercloud/gophercloud/openstack/compute/v2/extensions/aggregates"
+)
+
+func AggregateHosts() *schema.Table {
+	return &schema.Table{
+		Name:     "openstack_aggregate_hosts",
+		Resolver: fetchAggregateHosts,
+		Transform: transformers.TransformWithStruct(
+			&Single[string]{},
+			transformers.WithNameTransformer(transform.TagNameTransformer), // use cq-name tags to translate name
+			transformers.WithTypeTransformer(transform.TagTypeTransformer), // use cq-type tags to translate type
+		),
+	}
+}
+
+func fetchAggregateHosts(ctx context.Context, meta schema.ClientMeta, parent *schema.Resource, res chan<- interface{}) error {
+	api := meta.(*client.Client)
+	aggregate := parent.Item.(aggregates.Aggregate)
+	for _, v := range aggregate.Hosts {
+		host := &Single[string]{Name: v}
+		api.Logger.Debug().Int("aggregate id", aggregate.ID).Str("host", v).Msg("streaming aggregate host")
+		res <- host
+	}
+	return nil
+}

resources/aggregates.go

@@ -0,0 +1,62 @@
+package resources
+
+import (
+	"context"
+
+	"github.com/cloudquery/plugin-sdk/schema"
+	"github.com/cloudquery/plugin-sdk/transformers"
+	"github.com/dihedron/cq-plugin-utils/transform"
+	"github.com/dihedron/cq-source-openstack/client"
+	"github.com/gophercloud/gophercloud/openstack/compute/v2/extensions/aggregates"
+)
+
+func Aggregates() *schema.Table {
+	return &schema.Table{
+		Name:     "openstack_aggregates",
+		Resolver: fetchAggregates,
+		Transform: transformers.TransformWithStruct(
+			&aggregates.Aggregate{},
+			transformers.WithNameTransformer(transform.TagNameTransformer), // use cq-name tags to translate name
+			transformers.WithTypeTransformer(transform.TagTypeTransformer), // use cq-type tags to translate type
+			transformers.WithSkipFields("Links"),
+		),
+		Relations: []*schema.Table{
+			AggregateHosts(),
+		},
+	}
+}
+
+func fetchAggregates(ctx context.Context, meta schema.ClientMeta, parent *schema.Resource, res chan<- interface{}) error {
+
+	api := meta.(*client.Client)
+
+	compute, err := api.GetServiceClient(client.ComputeV2)
+	if err != nil {
+		api.Logger.Error().Err(err).Msg("error retrieving client")
+		return err
+	}
+
+	allPages, err := aggregates.List(compute).AllPages()
+	if err != nil {
+		api.Logger.Error().Err(err).Msg("error listing aggregates")
+		return err
+	}
+	allAggregates, err := aggregates.ExtractAggregates(allPages)
+	if err != nil {
+		api.Logger.Error().Err(err).Msg("error extracting aggregates")
+		return err
+	}
+	api.Logger.Debug().Int("count", len(allAggregates)).Msg("aggregates retrieved")
+
+	for _, aggregate := range allAggregates {
+		if ctx.Err() != nil {
+			api.Logger.Debug().Msg("context done, exit")
+			break
+		}
+		aggregate := aggregate
+		// api.Logger.Debug().Str("data", format.ToPrettyJSON(aggregate)).Msg("streaming aggregate")
+		api.Logger.Debug().Int("id", aggregate.ID).Msg("streaming aggregate")
+		res <- aggregate
+	}
+	return nil
+}

resources/hypervisors.go

@@ -0,0 +1,67 @@
+package resources
+
+import (
+	"context"
+
+	"github.com/cloudquery/plugin-sdk/schema"
+	"github.com/cloudquery/plugin-sdk/transformers"
+	"github.com/dihedron/cq-plugin-utils/format"
+	"github.com/dihedron/cq-plugin-utils/transform"
+	"github.com/dihedron/cq-source-openstack/client"
+	"github.com/gophercloud/gophercloud/openstack/compute/v2/extensions/hypervisors"
+)
+
+func Hypervisors() *schema.Table {
+	return &schema.Table{
+		Name:     "openstack_hypervisors",
+		Resolver: fetchHypervisors,
+		Transform: transformers.TransformWithStruct(
+			&hypervisors.Hypervisor{},
+			transformers.WithNameTransformer(transform.TagNameTransformer), // use cq-name tags to translate name
+			transformers.WithTypeTransformer(transform.TagTypeTransformer), // use cq-type tags to translate type
+			transformers.WithSkipFields("Links"),
+		),
+		Relations: []*schema.Table{
+			// ImageMetadata(),
+			// ImageProperties(),
+			// ImageTags(),
+		},
+	}
+}
+
+func fetchHypervisors(ctx context.Context, meta schema.ClientMeta, parent *schema.Resource, res chan<- interface{}) error {
+
+	api := meta.(*client.Client)
+
+	compute, err := api.GetServiceClient(client.ComputeV2)
+	if err != nil {
+		api.Logger.Error().Err(err).Msg("error retrieving client")
+		return err
+	}
+
+	opts := hypervisors.ListOpts{}
+
+	allPages, err := hypervisors.List(compute, opts).AllPages()
+	if err != nil {
+		api.Logger.Error().Err(err).Str("options", format.ToPrettyJSON(opts)).Msg("error listing hypervisors with options")
+		return err
+	}
+	allHypervisors, err := hypervisors.ExtractHypervisors(allPages)
+	if err != nil {
+		api.Logger.Error().Err(err).Msg("error extracting hypervisors")
+		return err
+	}
+	api.Logger.Debug().Int("count", len(allHypervisors)).Msg("hypervisors retrieved")
+
+	for _, hypervisor := range allHypervisors {
+		if ctx.Err() != nil {
+			api.Logger.Debug().Msg("context done, exit")
+			break
+		}
+		hypervisor := hypervisor
+		// api.Logger.Debug().Str("data", format.ToPrettyJSON(hypervisor)).Msg("streaming hypervisor")
+		api.Logger.Debug().Str("id", hypervisor.ID).Msg("streaming hypervisor")
+		res <- hypervisor
+	}
+	return nil
+}