add support for nsq

.dockerignore

@@ -1 +1,3 @@
 vendor
+docker-compose.yml
+docker-make.yml

Dockerfile.dev

@@ -0,0 +1,4 @@
+FROM alpine:3.8
+LABEL maintainer FI-TS Devops <devops@f-i-ts.de>
+COPY bin/metal-api /metal-api
+CMD ["/metal-api"]

Makefile

@@ -13,7 +13,7 @@ export GO111MODULE := on
 export CGO_ENABLED := 0
 
 
-.PHONY: all test up test-ci createmasterdata createtestdevices spec generate-client clean
+.PHONY: all test up test-ci createmasterdata createtestdevices spec generate-client clean vendor restart
 
 all: bin/$(BINARY);
 
@@ -31,6 +31,9 @@ clean:
 up:
 	docker-compose up --build
 
+vendor:
+	go mod vendor
+
 spec:
 	curl http://localhost:8080/apidocs.json >spec/metal-api.json
 
@@ -50,3 +53,7 @@ createmasterdata:
 
 createtestdevices:
 	@cat masterdata/testdevices.json | jq -r -c -M ".[]" | xargs -d'\n' -L1 -I'{}' curl -XPOST -H "Content-Type: application/json" -d '{}' http://localhost:8080/device/register
+
+restart:
+	docker build -t registry.fi-ts.io/metal/metal-api -f Dockerfile.dev .
+	docker-compose restart metal-api

cmd/metal-api/internal/datastore/datastore.go

@@ -29,9 +29,9 @@ type DeviceStore interface {
 	CreateDevice(device *metal.Device) error
 	DeleteDevice(id string) (*metal.Device, error)
 	UpdateDevice(oldDevice *metal.Device, newDevice *metal.Device) error
-	AllocateDevice(name string, description string, hostname string, projectid string, facilityid string, sizeid string, imageid string, sshPubKey string) (*metal.Device, error)
+	AllocateDevice(name string, description string, hostname string, projectid string, siteid string, sizeid string, imageid string, sshPubKey string) (*metal.Device, error)
 	FreeDevice(id string) (*metal.Device, error)
-	RegisterDevice(id string, facilityid string, hardware metal.DeviceHardware) (*metal.Device, error)
+	RegisterDevice(id string, siteid string, hardware metal.DeviceHardware) (*metal.Device, error)
 	Wait(id string, alloc Allocator) error
 }
 

cmd/metal-api/internal/datastore/rethinkstore/device.go

@@ -21,12 +21,12 @@ func (rs *RethinkStore) FindDevice(id string) (*metal.Device, error) {
 	if err != nil {
 		return nil, fmt.Errorf("cannot fetch results: %v", err)
 	}
-	if d.FacilityID != "" {
-		f, err := rs.FindFacility(d.FacilityID)
+	if d.SiteID != "" {
+		f, err := rs.FindFacility(d.SiteID)
 		if err != nil {
-			return nil, fmt.Errorf("illegal facility-id %q in device %q", d.FacilityID, id)
+			return nil, fmt.Errorf("illegal siteid %q in device %q", d.SiteID, id)
 		}
-		d.Facility = *f
+		d.Site = *f
 	}
 	if d.SizeID != "" {
 		s, err := rs.FindSize(d.SizeID)
@@ -100,8 +100,10 @@ func (rs *RethinkStore) CreateDevice(d *metal.Device) error {
 		d.ImageID = d.Image.ID
 	}
 	d.SizeID = d.Size.ID
-	d.FacilityID = d.Facility.ID
-	res, err := rs.deviceTable.Insert(d).RunWrite(rs.session)
+	d.SiteID = d.Site.ID
+	res, err := rs.deviceTable.Insert(d, r.InsertOpts{
+		Conflict: "replace",
+	}).RunWrite(rs.session)
 	if err != nil {
 		return fmt.Errorf("cannot create device in database: %v", err)
 	}
@@ -128,20 +130,20 @@ func (rs *RethinkStore) UpdateDevice(oldD *metal.Device, newD *metal.Device) err
 		return r.Branch(row.Field("changed").Eq(r.Expr(oldD.Changed)), newD, r.Error("the device was changed from another, please retry"))
 	}).RunWrite(rs.session)
 	if err != nil {
-		return fmt.Errorf("cannot update size: %v", err)
+		return fmt.Errorf("cannot update device: %v", err)
 	}
 	return nil
 }
 
-func (rs *RethinkStore) AllocateDevice(name string, description string, hostname string, projectid string, facilityid string, sizeid string, imageid string, sshPubKey string) (*metal.Device, error) {
+func (rs *RethinkStore) AllocateDevice(name string, description string, hostname string, projectid string, siteid string, sizeid string, imageid string, sshPubKey string) (*metal.Device, error) {
 	image, err := rs.FindImage(imageid)
 	if err != nil {
 		return nil, fmt.Errorf("image with id %q not found", imageid)
 	}
 	available, err := rs.waitTable.Filter(map[string]interface{}{
-		"project":    "",
-		"facilityid": facilityid,
-		"sizeid":     sizeid,
+		"project": "",
+		"siteid":  siteid,
+		"sizeid":  sizeid,
 	}).Run(rs.session)
 	if err != nil {
 		return nil, fmt.Errorf("cannot find free device: %v", err)
@@ -199,10 +201,10 @@ func (rs *RethinkStore) FreeDevice(id string) (*metal.Device, error) {
 	return device, nil
 }
 
-func (rs *RethinkStore) RegisterDevice(id string, facilityid string, hardware metal.DeviceHardware) (*metal.Device, error) {
-	fc, err := rs.FindFacility(facilityid)
+func (rs *RethinkStore) RegisterDevice(id string, siteid string, hardware metal.DeviceHardware) (*metal.Device, error) {
+	fc, err := rs.FindFacility(siteid)
 	if err != nil {
-		return nil, fmt.Errorf("facility with id %q not found", facilityid)
+		return nil, fmt.Errorf("facility with id %q not found", siteid)
 	}
 
 	sz := rs.determineSizeFromHardware(hardware)
@@ -212,7 +214,8 @@ func (rs *RethinkStore) RegisterDevice(id string, facilityid string, hardware me
 		device = &metal.Device{
 			ID:       id,
 			Size:     sz,
-			Facility: *fc,
+			Site:     *fc,
+			SiteID:   fc.ID,
 			Hardware: hardware,
 		}
 		err = rs.CreateDevice(device)
@@ -223,7 +226,8 @@ func (rs *RethinkStore) RegisterDevice(id string, facilityid string, hardware me
 	}
 	old := *device
 	device.Hardware = hardware
-	device.Facility = *fc
+	device.Site = *fc
+	device.SiteID = fc.ID
 	device.Size = sz
 
 	err = rs.UpdateDevice(&old, device)
@@ -313,7 +317,7 @@ func (rs *RethinkStore) fillDeviceList(data ...metal.Device) ([]metal.Device, er
 	res := make([]metal.Device, len(data), len(data))
 	for i, d := range data {
 		res[i] = d
-		res[i].Facility = facmap[d.FacilityID]
+		res[i].Site = facmap[d.SiteID]
 		size := szmap[d.SizeID]
 		res[i].Size = &size
 		if d.ImageID != "" {

cmd/metal-api/internal/service/device-service.go

@@ -12,7 +12,7 @@ import (
 	nbdevice "git.f-i-ts.de/cloud-native/maas/metal-api/netbox-api/client/device"
 	"git.f-i-ts.de/cloud-native/maas/metal-api/netbox-api/models"
 	"git.f-i-ts.de/cloud-native/maas/metal-api/pkg/metal"
-	"git.f-i-ts.de/cloud-native/maas/metal-api/pkg/mq"
+	"git.f-i-ts.de/cloud-native/metallib/bus"
 	restful "github.com/emicklei/go-restful"
 	restfulspec "github.com/emicklei/go-restful-openapi"
 	"github.com/inconshreveable/log15"
@@ -24,7 +24,7 @@ const (
 
 type deviceResource struct {
 	log15.Logger
-	*mq.Publisher
+	*bus.Publisher
 	netbox *client.NetboxAPIProxy
 	ds     datastore.Datastore
 }
@@ -34,7 +34,7 @@ type allocateRequest struct {
 	Hostname    string `json:"hostname" description:"the hostname for the allocated device"`
 	Description string `json:"description" description:"the description for the allocated device" optional:"true"`
 	ProjectID   string `json:"projectid" description:"the project id to assign this device to"`
-	FacilityID  string `json:"facilityid" description:"the facility id to assign this device to"`
+	SiteID      string `json:"siteid" description:"the site id to assign this device to"`
 	SizeID      string `json:"sizeid" description:"the size id to assign this device to"`
 	ImageID     string `json:"imageid" description:"the image id to assign this device to"`
 	SSHPubKey   string `json:"ssh_pub_key" description:"the public ssh key to access the device with"`
@@ -50,7 +50,7 @@ type registerRequest struct {
 func NewDevice(
 	log log15.Logger,
 	ds datastore.Datastore,
-	pub *mq.Publisher,
+	pub *bus.Publisher,
 	netbox *client.NetboxAPIProxy) *restful.WebService {
 	dr := deviceResource{
 		Logger:    log,
@@ -224,7 +224,7 @@ func (dr deviceResource) allocateDevice(request *restful.Request, response *rest
 		sendError(dr, response, "allocateDevice", http.StatusInternalServerError, fmt.Errorf("Cannot read request: %v", err))
 		return
 	}
-	d, err := dr.ds.AllocateDevice(allocate.Name, allocate.Description, allocate.Hostname, allocate.ProjectID, allocate.FacilityID, allocate.SizeID, allocate.ImageID, allocate.SSHPubKey)
+	d, err := dr.ds.AllocateDevice(allocate.Name, allocate.Description, allocate.Hostname, allocate.ProjectID, allocate.SiteID, allocate.SizeID, allocate.ImageID, allocate.SSHPubKey)
 	if err != nil {
 		if err == datastore.ErrNoDeviceAvailable {
 			sendError(dr, response, "allocateDevice", http.StatusNotFound, err)

cmd/metal-api/main.go

@@ -2,8 +2,6 @@ package main
 
 import (
 	"fmt"
-	"git.f-i-ts.de/cloud-native/maas/metal-api/netbox-api/client"
-	"github.com/go-openapi/strfmt"
 	"net/http"
 	"os"
 	"os/signal"
@@ -14,11 +12,15 @@ import (
 	"git.f-i-ts.de/cloud-native/maas/metal-api/cmd/metal-api/internal/datastore/rethinkstore"
 	"git.f-i-ts.de/cloud-native/maas/metal-api/cmd/metal-api/internal/service"
 	"git.f-i-ts.de/cloud-native/maas/metal-api/cmd/metal-api/internal/utils"
+	"git.f-i-ts.de/cloud-native/maas/metal-api/netbox-api/client"
 	"git.f-i-ts.de/cloud-native/maas/metal-api/pkg/health"
-	"git.f-i-ts.de/cloud-native/maas/metal-api/pkg/mq"
+	"git.f-i-ts.de/cloud-native/maas/metal-api/pkg/metal"
+	"git.f-i-ts.de/cloud-native/metallib/bus"
+	"git.f-i-ts.de/cloud-native/metallib/zapup"
 	restful "github.com/emicklei/go-restful"
 	restfulspec "github.com/emicklei/go-restful-openapi"
 	"github.com/go-openapi/spec"
+	"github.com/go-openapi/strfmt"
 	"github.com/inconshreveable/log15"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -35,7 +37,7 @@ var (
 	builddate string
 	cfgFile   string
 	ds        datastore.Datastore
-	producer  *mq.Publisher
+	producer  *bus.Publisher
 	netbox    *client.NetboxAPIProxy
 	logger    log15.Logger
 	debug     = false
@@ -80,6 +82,7 @@ func init() {
 	rootCmd.Flags().StringP("db-password", "", "", "the database password to use")
 
 	rootCmd.Flags().StringP("nsqd-addr", "", "nsqd:4150", "the address of the nsqd")
+	rootCmd.Flags().StringP("nsqd-http-addr", "", "nsqd:4151", "the address of the nsqd rest endpoint")
 	rootCmd.Flags().StringP("nsqlookupd-addr", "", "nsqlookupd:4160", "the address of the nsqlookupd as a commalist")
 
 	rootCmd.Flags().StringP("netbox-addr", "", "localhost:8001", "the address of netbox proxy")
@@ -185,13 +188,15 @@ func initNetboxProxy() {
 
 func initEventBus() {
 	nsqd := viper.GetString("nsqd-addr")
-	lookupds := viper.GetString("nsqlookupd-addr")
-	client := mq.NewClient(strings.Split(lookupds, ","))
-	p, err := client.Producer(nsqd)
+	httpnsqd := viper.GetString("nsqd-http-addr")
+	p, err := bus.NewPublisher(zapup.MustRootLogger(), nsqd, httpnsqd)
 	if err != nil {
 		panic(err)
 	}
-	log15.Info("nsq connected", "nsqd", nsqd, "lookupds", lookupds)
+	log15.Info("nsq connected", "nsqd", nsqd)
+	if err := p.CreateTopic(string(metal.TopicDevice)); err != nil {
+		panic(err)
+	}
 	producer = p
 }
 

docker-compose.yml

@@ -4,11 +4,16 @@ networks:
   appl:
     driver: bridge
 
+volumes:
+  nsq-data-volume:
+
 services:
   nsqlookupd:
     image: nsqio/nsq:v1.1.0
     networks:
       - appl
+    volumes:
+      - nsq-data-volume:/data
     command: /nsqlookupd -log-level error
     ports:
       - "4160"
@@ -66,6 +71,20 @@ services:
       METAL_API_DB_ADDR: rethinkdb:28015
       # we need a better solution for this. establish a docker-network? use "host_mode" for anything?
       METAL_API_NETBOX_ADDR: 172.17.0.1:8001
-      METAL_API_NSQ_ADDR: nsqd:4150
-      METAL_API_NSQLOOKUPD_ADDR: nsqlookupd:4160
+      METAL_API_NSQD_ADDR: nsqd:4150
+      METAL_API_NSQD_HTTP_ADDR: nsqd:4151
       METAL_API_LOG_LEVEL: "debug"
+
+  metal-core:
+    networks:
+      - appl
+    image: registry.fi-ts.io/metal/metal-core
+    container_name: metal-core
+    environment:
+      METAL_CORE_MQ_ADDRESS: nsqlookupd:4161
+      METAL_CORE_CONTROL_PLANE_IP: localhost
+      METAL_CORE_SITE_ID: FRA
+      METAL_CORE_RACK_ID: Vagrant Rack 1
+      METAL_CORE_LOG_LEVEL: DEBUG
+      METAL_CORE_METAL_API_PORT: 8090
+      METAL_CORE_HAMMER_IMAGE_PREFIX: metal-hammer

go.mod

@@ -1,6 +1,7 @@
 module git.f-i-ts.de/cloud-native/maas/metal-api
 
 require (
+	git.f-i-ts.de/cloud-native/metallib v0.0.0-20181026124130-dac2b6e8a196
 	github.com/emicklei/go-restful v2.8.0+incompatible
 	github.com/emicklei/go-restful-openapi v1.0.0
 	github.com/go-openapi/errors v0.17.0
@@ -10,7 +11,6 @@ require (
 	github.com/go-openapi/swag v0.17.0
 	github.com/go-openapi/validate v0.17.0
 	github.com/go-stack/stack v1.8.0
-	github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db // indirect
 	github.com/inconshreveable/log15 v0.0.0-20180818164646-67afb5ed74ec
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.5 // indirect
@@ -25,3 +25,5 @@ require (
 	golang.org/x/net v0.0.0-20181005035420-146acd28ed58
 	gopkg.in/rethinkdb/rethinkdb-go.v5 v5.0.1
 )
+
+// replace git.f-i-ts.de/cloud-native/metallib => ../../metallib

go.sum

@@ -1,3 +1,9 @@
+git.f-i-ts.de/cloud-native/metallib v0.0.0-20181023072015-c87005a4a3a6 h1:L+wNftiT/d2jN2Ce5dLV5t09nkZ6cUSuGcUg4O6B+Zs=
+git.f-i-ts.de/cloud-native/metallib v0.0.0-20181023072015-c87005a4a3a6/go.mod h1:aGWPjTVphlLwGrWdvtsHJcTNnNYIkIddK2Li/6HzgO8=
+git.f-i-ts.de/cloud-native/metallib v0.0.0-20181023083140-89a635bff502 h1:atPXR6M0JIMbcVfk+m8rgPAjFX9nIoEZzlT4jKZmtOA=
+git.f-i-ts.de/cloud-native/metallib v0.0.0-20181023083140-89a635bff502/go.mod h1:aGWPjTVphlLwGrWdvtsHJcTNnNYIkIddK2Li/6HzgO8=
+git.f-i-ts.de/cloud-native/metallib v0.0.0-20181026124130-dac2b6e8a196 h1:sP6k1hcVt939VGhme2NuC6GCsCQslwm6U6Nj2CmuO9g=
+git.f-i-ts.de/cloud-native/metallib v0.0.0-20181026124130-dac2b6e8a196/go.mod h1:esbYGe5VRpmcDTI0LBY5ZsnFXLqSLlmoodTWBS0LcC4=
 github.com/PuerkitoBio/purell v1.1.0 h1:rmGxhojJlM0tuKtfdvliR84CFHljx9ag64t2xmVkjK4=
 github.com/PuerkitoBio/purell v1.1.0/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
 github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 h1:d+Bc7a5rLufV/sSk/8dngufqelfh6jnri85riMAaF/M=
@@ -109,6 +115,12 @@ github.com/spf13/viper v1.2.0/go.mod h1:P4AexN0a+C9tGAnUFNwDMYYZv3pjFuvmeiMyKRaN
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+go.uber.org/atomic v1.3.2 h1:2Oa65PReHzfn29GpvgsYwloV9AVFHPDk8tYxt2c2tr4=
+go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
+go.uber.org/multierr v1.1.0 h1:HoEmRHQPVSqub6w2z2d2EOVs2fjyFRGyofhKuyDq0QI=
+go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
+go.uber.org/zap v1.9.1 h1:XCJQEf3W6eZaVwhRBof6ImoYGJSITeKWsyeh3HFu/5o=
+go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 golang.org/x/crypto v0.0.0-20180820150726-614d502a4dac h1:7d7lG9fHOLdL6jZPtnV4LpI41SbohIJ1Atq7U991dMg=
 golang.org/x/crypto v0.0.0-20180820150726-614d502a4dac/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=

pkg/metal/device.go

@@ -9,8 +9,8 @@ type Device struct {
 	Created     time.Time      `json:"created" description:"the creation time of this machine" optional:"true" readOnly:"true" rethinkdb:"created"`
 	Changed     time.Time      `json:"changed" description:"the last changed timestamp" optional:"true" readOnly:"true" rethinkdb:"changed"`
 	Project     string         `json:"project" description:"the project that this device is assigned to" rethinkdb:"project"`
-	Facility    Facility       `json:"facility" description:"the facility assigned to this device" readOnly:"true" rethinkdb:"-"`
-	FacilityID  string         `json:"-" rethinkdb:"facilityid"`
+	Site        Facility       `json:"site" description:"the site assigned to this device" readOnly:"true" rethinkdb:"-"`
+	SiteID      string         `json:"-" rethinkdb:"siteid"`
 	Image       *Image         `json:"image" description:"the image assigned to this device" readOnly:"true"  rethinkdb:"-"`
 	ImageID     string         `json:"-" rethinkdb:"imageid"`
 	Size        *Size          `json:"size" description:"the size of this device" readOnly:"true" rethinkdb:"-"`

pkg/metal/metal.go

@@ -2,9 +2,19 @@ package metal
 
 type EventType string
 
-// Some EventType enums.
+type NSQTopic string
+
+// Some enums.
 const (
 	CREATE EventType = "create"
 	UPDATE EventType = "update"
 	DELETE EventType = "delete"
+
+	TopicDevice NSQTopic = "device"
+)
+
+var (
+	Topics = []NSQTopic{
+		TopicDevice,
+	}
 )