Merge pull request #216 from cloudflare/mapper-refactor

Mapper package refactoring

internal/alertmanager/dedup_test.go

@@ -17,7 +17,10 @@ func init() {
 	log.SetLevel(log.ErrorLevel)
 	for i, uri := range mock.ListAllMockURIs() {
 		name := fmt.Sprintf("dedup-mock-%d", i)
-		am := alertmanager.NewAlertmanager(name, uri, alertmanager.WithRequestTimeout(time.Second))
+		am, err := alertmanager.NewAlertmanager(name, uri, alertmanager.WithRequestTimeout(time.Second))
+		if err != nil {
+			log.Fatal(err)
+		}
 		alertmanager.RegisterAlertmanager(am)
 	}
 }

internal/alertmanager/models.go

@@ -1,6 +1,7 @@
 package alertmanager
 
 import (
+	"encoding/json"
 	"fmt"
 	"path"
 	"sort"
@@ -34,6 +35,8 @@ type Alertmanager struct {
 	Name           string        `json:"name"`
 	// whenever this instance should be proxied
 	ProxyRequests bool
+	// transport instances are specific to URI scheme we collect from
+	transport transport.Transport
 	// lock protects data access while updating
 	lock sync.RWMutex
 	// fields for storing pulled data
@@ -56,9 +59,19 @@ func (am *Alertmanager) detectVersion() string {
 		return defaultVersion
 	}
 	ver := alertmanagerVersion{}
-	err = transport.ReadJSON(url, am.RequestTimeout, &ver)
+
+	// read raw body from the source
+	source, err := am.transport.Read(url)
+	defer source.Close()
+	if err != nil {
+		log.Errorf("[%s] %s request failed: %s", am.Name, url, err)
+		return defaultVersion
+	}
+
+	// decode body as JSON
+	err = json.NewDecoder(source).Decode(&ver)
 	if err != nil {
-		log.Errorf("[%s] %s request failed: %s", am.Name, url, err.Error())
+		log.Errorf("[%s] %s failed to decode as JSON: %s", am.Name, url, err)
 		return defaultVersion
 	}
 
@@ -91,8 +104,24 @@ func (am *Alertmanager) pullSilences(version string) error {
 		return err
 	}
 
+	// generate full URL to collect silences from
+	url, err := mapper.AbsoluteURL(am.URI)
+	if err != nil {
+		log.Errorf("[%s] Failed to generate silences endpoint URL: %s", am.Name, err)
+		return err
+	}
+
 	start := time.Now()
-	silences, err := mapper.GetSilences(am.URI, am.RequestTimeout)
+	// read raw body from the source
+	source, err := am.transport.Read(url)
+	defer source.Close()
+	if err != nil {
+		log.Errorf("[%s] %s request failed: %s", am.Name, url, err)
+		return err
+	}
+
+	// decode body text
+	silences, err := mapper.Decode(source)
 	if err != nil {
 		return err
 	}
@@ -134,8 +163,24 @@ func (am *Alertmanager) pullAlerts(version string) error {
 		return err
 	}
 
+	// generate full URL to collect alerts from
+	url, err := mapper.AbsoluteURL(am.URI)
+	if err != nil {
+		log.Errorf("[%s] Failed to generate alerts endpoint URL: %s", am.Name, err)
+		return err
+	}
+
 	start := time.Now()
-	groups, err := mapper.GetAlerts(am.URI, am.RequestTimeout)
+	// read raw body from the source
+	source, err := am.transport.Read(url)
+	defer source.Close()
+	if err != nil {
+		log.Errorf("[%s] %s request failed: %s", am.Name, url, err)
+		return err
+	}
+
+	// decode body text
+	groups, err := mapper.Decode(source)
 	if err != nil {
 		return err
 	}

internal/alertmanager/upstream.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/cloudflare/unsee/internal/models"
+	"github.com/cloudflare/unsee/internal/transport"
 
 	log "github.com/sirupsen/logrus"
 )
@@ -18,7 +19,7 @@ var (
 )
 
 // NewAlertmanager creates a new Alertmanager instance
-func NewAlertmanager(name, uri string, opts ...Option) *Alertmanager {
+func NewAlertmanager(name, uri string, opts ...Option) (*Alertmanager, error) {
 	am := &Alertmanager{
 		URI:            uri,
 		RequestTimeout: time.Second * 10,
@@ -40,7 +41,13 @@ func NewAlertmanager(name, uri string, opts ...Option) *Alertmanager {
 		opt(am)
 	}
 
-	return am
+	var err error
+	am.transport, err = transport.NewTransport(am.URI, am.RequestTimeout)
+	if err != nil {
+		return am, err
+	}
+
+	return am, nil
 }
 
 // RegisterAlertmanager will add an Alertmanager instance to the list of

internal/alertmanager/version.go

@@ -1,6 +1,7 @@
 package alertmanager
 
 import (
+	"encoding/json"
 	"time"
 
 	"github.com/cloudflare/unsee/internal/transport"
@@ -30,7 +31,15 @@ func GetVersion(uri string, timeout time.Duration) string {
 		return defaultVersion
 	}
 	ver := alertmanagerVersion{}
-	err = transport.ReadJSON(url, timeout, &ver)
+
+	t, err := transport.NewTransport(uri, timeout)
+	if err != nil {
+		log.Errorf("Unable to get the version information from %s", url)
+		return defaultVersion
+	}
+
+	source, err := t.Read(url)
+	err = json.NewDecoder(source).Decode(&ver)
 	if err != nil {
 		log.Errorf("%s request failed: %s", url, err.Error())
 		return defaultVersion

internal/filters/filter_test.go

@@ -485,7 +485,10 @@ var tests = []filterTest{
 func TestFilters(t *testing.T) {
 	log.SetLevel(log.ErrorLevel)
 
-	am := alertmanager.NewAlertmanager("test", "http://localhost", alertmanager.WithRequestTimeout(time.Second))
+	am, err := alertmanager.NewAlertmanager("test", "http://localhost", alertmanager.WithRequestTimeout(time.Second))
+	if err != nil {
+		t.Error(err)
+	}
 	for _, ft := range tests {
 		alert := models.Alert(ft.Alert)
 		if &ft.Silence != nil {

internal/mapper/mapper.go

@@ -2,7 +2,7 @@ package mapper
 
 import (
 	"fmt"
-	"time"
+	"io"
 
 	"github.com/cloudflare/unsee/internal/models"
 )
@@ -12,11 +12,22 @@ var (
 	silenceMappers = []SilenceMapper{}
 )
 
-// AlertMapper implements Alertmanager -> unsee alert data mapping that works
-// for a specific range of Alertmanager versions
-type AlertMapper interface {
+// Mapper converts Alertmanager response body and maps to unsee data structures
+type Mapper interface {
 	IsSupported(version string) bool
-	GetAlerts(uri string, timeout time.Duration) ([]models.AlertGroup, error)
+	AbsoluteURL(baseURI string) (string, error)
+}
+
+// AlertMapper handles mapping of Alertmanager alert information to unsee AlertGroup models
+type AlertMapper interface {
+	Mapper
+	Decode(io.ReadCloser) ([]models.AlertGroup, error)
+}
+
+// SilenceMapper handles mapping of Alertmanager silence information to unsee Silence models
+type SilenceMapper interface {
+	Mapper
+	Decode(io.ReadCloser) ([]models.Silence, error)
 }
 
 // RegisterAlertMapper allows to register mapper implementing alert data
@@ -35,14 +46,6 @@ func GetAlertMapper(version string) (AlertMapper, error) {
 	return nil, fmt.Errorf("Can't find alert mapper for Alertmanager %s", version)
 }
 
-// SilenceMapper implements Alertmanager -> unsee silence data mapping that
-// works for a specific range of Alertmanager versions
-type SilenceMapper interface {
-	Release() string
-	IsSupported(version string) bool
-	GetSilences(uri string, timeout time.Duration) ([]models.Silence, error)
-}
-
 // RegisterSilenceMapper allows to register mapper implementing silence data
 // handling for specific Alertmanager versions
 func RegisterSilenceMapper(m SilenceMapper) {

internal/mapper/v04/alerts.go

@@ -5,7 +5,9 @@
 package v04
 
 import (
+	"encoding/json"
 	"errors"
+	"io"
 	"sort"
 	"strconv"
 	"time"
@@ -52,25 +54,25 @@ type AlertMapper struct {
 	mapper.AlertMapper
 }
 
+// AbsoluteURL for alerts API endpoint this mapper supports
+func (m AlertMapper) AbsoluteURL(baseURI string) (string, error) {
+	return transport.JoinURL(baseURI, "api/v1/alerts/groups")
+}
+
 // IsSupported returns true if given version string is supported
 func (m AlertMapper) IsSupported(version string) bool {
 	versionRange := semver.MustParseRange(">=0.4.0 <0.5.0")
 	return versionRange(semver.MustParse(version))
 }
 
-// GetAlerts will make a request to Alertmanager API and parse the response
-// It will only return alerts or error (if any)
-func (m AlertMapper) GetAlerts(uri string, timeout time.Duration) ([]models.AlertGroup, error) {
+// Decode Alertmanager API response body and return unsee model instances
+func (m AlertMapper) Decode(source io.ReadCloser) ([]models.AlertGroup, error) {
 	groups := []models.AlertGroup{}
 	receivers := map[string]alertsGroupReceiver{}
 	resp := alertsGroupsAPISchema{}
 
-	url, err := transport.JoinURL(uri, "api/v1/alerts/groups")
-	if err != nil {
-		return groups, err
-	}
-
-	err = transport.ReadJSON(url, timeout, &resp)
+	defer source.Close()
+	err := json.NewDecoder(source).Decode(&resp)
 	if err != nil {
 		return groups, err
 	}

internal/mapper/v04/silences.go

@@ -5,9 +5,9 @@
 package v04
 
 import (
+	"encoding/json"
 	"errors"
-	"fmt"
-	"math"
+	"io"
 	"strconv"
 	"time"
 
@@ -47,26 +47,24 @@ type SilenceMapper struct {
 	mapper.SilenceMapper
 }
 
+// AbsoluteURL for silences API endpoint this mapper supports
+func (m SilenceMapper) AbsoluteURL(baseURI string) (string, error) {
+	return transport.JoinURL(baseURI, "api/v1/silences")
+}
+
 // IsSupported returns true if given version string is supported
 func (m SilenceMapper) IsSupported(version string) bool {
 	versionRange := semver.MustParseRange(">=0.4.0 <0.5.0")
 	return versionRange(semver.MustParse(version))
 }
 
-// GetSilences will make a request to Alertmanager API and parse the response
-// It will only return silences or error (if any)
-func (m SilenceMapper) GetSilences(uri string, timeout time.Duration) ([]models.Silence, error) {
+// Decode Alertmanager API response body and return unsee model instances
+func (m SilenceMapper) Decode(source io.ReadCloser) ([]models.Silence, error) {
 	silences := []models.Silence{}
 	resp := silenceAPISchema{}
 
-	url, err := transport.JoinURL(uri, "api/v1/silences")
-	if err != nil {
-		return silences, err
-	}
-
-	// Alertmanager 0.4 uses pagination for silences
-	url = fmt.Sprintf("%s?limit=%d", url, math.MaxUint32)
-	err = transport.ReadJSON(url, timeout, &resp)
+	defer source.Close()
+	err := json.NewDecoder(source).Decode(&resp)
 	if err != nil {
 		return silences, err
 	}

internal/mapper/v05/alerts.go

@@ -5,7 +5,9 @@
 package v05
 
 import (
+	"encoding/json"
 	"errors"
+	"io"
 	"sort"
 	"time"
 
@@ -51,25 +53,25 @@ type AlertMapper struct {
 	mapper.AlertMapper
 }
 
+// AbsoluteURL for alerts API endpoint this mapper supports
+func (m AlertMapper) AbsoluteURL(baseURI string) (string, error) {
+	return transport.JoinURL(baseURI, "api/v1/alerts/groups")
+}
+
 // IsSupported returns true if given version string is supported
 func (m AlertMapper) IsSupported(version string) bool {
 	versionRange := semver.MustParseRange(">=0.5.0  <=0.6.0")
 	return versionRange(semver.MustParse(version))
 }
 
-// GetAlerts will make a request to Alertmanager API and parse the response
-// It will only return alerts or error (if any)
-func (m AlertMapper) GetAlerts(uri string, timeout time.Duration) ([]models.AlertGroup, error) {
+// Decode Alertmanager API response body and return unsee model instances
+func (m AlertMapper) Decode(source io.ReadCloser) ([]models.AlertGroup, error) {
 	groups := []models.AlertGroup{}
 	receivers := map[string]alertsGroupReceiver{}
 	resp := alertsGroupsAPISchema{}
 
-	url, err := transport.JoinURL(uri, "api/v1/alerts/groups")
-	if err != nil {
-		return groups, err
-	}
-
-	err = transport.ReadJSON(url, timeout, &resp)
+	defer source.Close()
+	err := json.NewDecoder(source).Decode(&resp)
 	if err != nil {
 		return groups, err
 	}