Merge pull request #34 from ARGOeu/devel

Version 1.0.1

Jenkinsfile

@@ -63,9 +63,6 @@ pipeline {
         }
         success {
             script{
-                if ( env.BRANCH_NAME == 'devel' ) {
-                    build job: '/ARGO-utils/argo-swagger-docs', propagate: false
-                }
                 if ( env.BRANCH_NAME == 'master' || env.BRANCH_NAME == 'devel' ) {
                     slackSend( message: ":rocket: New version for <$BUILD_URL|$PROJECT_DIR>:$BRANCH_NAME Job: $JOB_NAME !")
                 }

README.md

@@ -68,7 +68,8 @@ Before you start, you need to issue a valid certificate.
   "trust_unknown_cas": false,
   "log_level": "INFO",
   "skip_subs_load": false,
-  "acl": ["OU=my.local,O=mkcert development certificate"]
+  "acl": ["OU=my.local,O=mkcert development certificate"],
+  "syslog_enabled": false
 }
  ```
  - `service_port:` The port that the service will bind to.  
@@ -101,6 +102,8 @@ Before you start, you need to issue a valid certificate.
  `ams_token`). You can control this behavior and decide whether or not to pre-load any already active subscriptions.
 
  - `acl`: List of certificate DNs which are allowed to access the service.
+
+ -  `syslog_enabled`: Direct logging of the service to the syslog socket
 
 You can find the configuration template at `conf/ams-push-server-config.template`.
 ## Managing the protocol buffers and gRPC definitions

ams-push-server.spec

@@ -3,7 +3,7 @@
 
 Name: ams-push-server
 Summary: ARGO Ams Push Server.
-Version: 1.0.0
+Version: 1.0.1
 Release: 1%{?dist}
 License: ASL 2.0
 Buildroot: %{_tmppath}/%{name}-buildroot
@@ -57,5 +57,7 @@ go clean
 %attr(0644,root,root) /usr/lib/systemd/system/ams-push-server.service
 
 %changelog
+* Tue Oct 5 2021 Agelos Tsalapatis  <agelos.tsal@gmail.com> 1.0.1-1%{?dist}
+- Release of ams-push-server 1.0.1
 * Wed May 27 2020 Agelos Tsalapatis  <agelos.tsal@gmail.com> 1.0.0-1%{?dist}
 - Release of ams-push-server 1.0.0

api/v1/grpc/proto/ams.proto

@@ -75,6 +75,8 @@ message PushConfig {
     int64 max_messages = 3;
     // Required. Retry policy.
     RetryPolicy retry_policy = 2;
+    // Required. Authorization header that the sent messages should include into the request
+    string authorization_header = 4;
 }
 
 // RetryPolicy holds information regarding the retry policy.

api/v1/grpc/server.go

@@ -52,7 +52,8 @@ func NewPushService(cfg *config.Config) *PushService {
 	// build the client
 	transCfg := &http.Transport{
 		TLSClientConfig: &tls.Config{
-			InsecureSkipVerify: !ps.Cfg.VerifySSL},
+			InsecureSkipVerify: !ps.Cfg.VerifySSL,
+		},
 	}
 
 	client := &http.Client{
@@ -144,7 +145,7 @@ func (ps *PushService) ActivateSubscription(ctx context.Context, r *amsPb.Activa
 	c, _ := consumers.New(consumers.AmsHttpConsumerType, r.Subscription.FullName, ps.Cfg, ps.Client)
 
 	// choose a sender
-	s, _ := senders.New(senders.HttpSenderType, r.Subscription.PushConfig.PushEndpoint, ps.Client)
+	s, _ := senders.New(senders.HttpSenderType, *r.Subscription.PushConfig, ps.Client)
 
 	worker, err := push.New(r.Subscription, c, s, ps.deactivateChan)
 	if err != nil {
@@ -298,8 +299,9 @@ func (ps *PushService) loadSubscriptions() {
 						FullName:  sub.FullName,
 						FullTopic: sub.FullTopic,
 						PushConfig: &amsPb.PushConfig{
-							PushEndpoint: sub.PushCfg.Pend,
-							MaxMessages:  sub.PushCfg.MaxMessages,
+							PushEndpoint:        sub.PushCfg.Pend,
+							AuthorizationHeader: sub.PushCfg.AuthorizationHeader.Value,
+							MaxMessages:         sub.PushCfg.MaxMessages,
 							RetryPolicy: &amsPb.RetryPolicy{
 								Period: sub.PushCfg.RetPol.Period,
 								Type:   sub.PushCfg.RetPol.PolicyType,
@@ -436,9 +438,15 @@ type Subscription struct {
 
 // PushConfig holds optional configuration for push operations
 type PushConfig struct {
-	Pend        string      `json:"pushEndpoint"`
-	MaxMessages int64       `json:"maxMessages"`
-	RetPol      RetryPolicy `json:"retryPolicy"`
+	Pend                string              `json:"pushEndpoint"`
+	AuthorizationHeader AuthorizationHeader `json:"authorization_header"`
+	MaxMessages         int64               `json:"maxMessages"`
+	RetPol              RetryPolicy         `json:"retryPolicy"`
+}
+
+// AuthorizationHeader holds an optional value to be supplied as an Authorization header to push requests
+type AuthorizationHeader struct {
+	Value string `json:"value"`
 }
 
 // RetryPolicy holds information on retry policies

api/v1/grpc/server_test.go

@@ -288,9 +288,14 @@ func (suite *ServerTestSuite) TestGetSubscription() {
 		Period:     300,
 	}
 
+	authz := AuthorizationHeader{
+		Value: "auth-header-1",
+	}
+
 	pc := PushConfig{
-		Pend:   "example.com:9999",
-		RetPol: rp,
+		Pend:                "example.com:9999",
+		AuthorizationHeader: authz,
+		RetPol:              rp,
 	}
 
 	expectedSub := Subscription{
@@ -397,10 +402,14 @@ func (m *MockRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) {
 			PolicyType: "linear",
 			Period:     300,
 		}
+		authz := AuthorizationHeader{
+			Value: "auth-header-1",
+		}
 
 		pc := PushConfig{
-			Pend:   "example.com:9999",
-			RetPol: rp,
+			Pend:                "example.com:9999",
+			AuthorizationHeader: authz,
+			RetPol:              rp,
 		}
 
 		s := Subscription{

conf/ams-push-server-config.template

@@ -11,5 +11,6 @@
   "trust_unknown_cas": false,
   "log_level": "INFO",
   "skip_subs_load": false,
-  "acl": ["OU=my.local,O=mkcert development certificate"]
+  "acl": ["OU=my.local,O=mkcert development certificate"],
+  "syslog_enabled": false
 }

config/config.go

@@ -6,8 +6,10 @@ import (
 	"encoding/json"
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
+	lSyslog "github.com/sirupsen/logrus/hooks/syslog"
 	"io"
 	"io/ioutil"
+	"log/syslog"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -44,6 +46,8 @@ type Config struct {
 	tlsConfig *tls.Config
 	// list of certificate DNs that should be allowed to access the service
 	ACL []string `json:"acl"`
+	// Enable direct logging of the service to the syslog facility
+	SyslogEnabled bool `json:"syslog_enabled"`
 }
 
 var logLevels = map[string]log.Level{
@@ -109,6 +113,13 @@ func (cfg *Config) LoadFromJson(from io.Reader) error {
 			continue
 		}
 
+		if cfg.SyslogEnabled {
+			hook, err := lSyslog.NewSyslogHook("", "", syslog.LOG_INFO, "")
+			if err == nil {
+				log.AddHook(hook)
+			}
+		}
+
 		log.WithFields(
 			log.Fields{
 				"type":  "service_log",

config/config_test.go

@@ -36,6 +36,7 @@ func (suite *ConfigTestSuite) TestValidateRequired() {
 		LogLevel:                  "INFO",
 		SkipSubsLoad:              true,
 		ACL:                       []string{"OU=my.local,O=mkcert development certificate"},
+		SyslogEnabled:             true,
 	}
 
 	// test the case where where everything is set properly
@@ -59,7 +60,8 @@ func (suite *ConfigTestSuite) TestLoadFromJson() {
   "trust_unknown_cas": true,
   "log_level": "INFO",
   "skip_subs_load": true,
-  "acl": ["OU=my.local,O=mkcert development certificate"]
+  "acl": ["OU=my.local,O=mkcert development certificate"],
+  "syslog_enabled": true
 }
 `
 	cfg := new(Config)
@@ -79,6 +81,7 @@ func (suite *ConfigTestSuite) TestLoadFromJson() {
 	suite.Equal("INFO", cfg.LogLevel)
 	suite.Equal(true, cfg.SkipSubsLoad)
 	suite.Equal([]string{"OU=my.local,O=mkcert development certificate"}, cfg.ACL)
+	suite.Equal(true, cfg.SyslogEnabled)
 
 	suite.Nil(e1)
 

main.go

@@ -7,22 +7,13 @@ import (
 	amsgRPC "github.com/ARGOeu/ams-push-server/api/v1/grpc"
 	"github.com/ARGOeu/ams-push-server/config"
 	log "github.com/sirupsen/logrus"
-	lSyslog "github.com/sirupsen/logrus/hooks/syslog"
 	"io/ioutil"
-	"log/syslog"
 	"net"
 )
 
 func init() {
-
 	fmter := &log.TextFormatter{FullTimestamp: true, DisableColors: true}
-	hook, err := lSyslog.NewSyslogHook("", "", syslog.LOG_DEBUG, "")
-
 	log.SetFormatter(fmter)
-
-	if err == nil {
-		log.AddHook(hook)
-	}
 }
 
 func main() {

push/worker_test.go

@@ -21,7 +21,7 @@ type WorkerTestSuite struct {
 func (suite *WorkerTestSuite) TestNew() {
 
 	c := consumers.NewAmsHttpConsumer("", "", "", &http.Client{})
-	s := senders.NewHttpSender("", &http.Client{})
+	s := senders.NewHttpSender("", "", &http.Client{})
 	sub := &amsPb.Subscription{
 		PushConfig: &amsPb.PushConfig{
 			MaxMessages: 1,

senders/httpsender.go

@@ -14,15 +14,17 @@ const ApplicationJson = "application/json"
 
 // HttpSender delivers data to any http endpoint
 type HttpSender struct {
-	client   *http.Client
-	endpoint string
+	client      *http.Client
+	endpoint    string
+	authZHeader string
 }
 
 // NewHttpSender initialises and returns a new http sender
-func NewHttpSender(endpoint string, client *http.Client) *HttpSender {
+func NewHttpSender(endpoint, authz string, client *http.Client) *HttpSender {
 	s := new(HttpSender)
 	s.client = client
 	s.endpoint = endpoint
+	s.authZHeader = authz
 	return s
 }
 
@@ -50,6 +52,9 @@ func (s *HttpSender) Send(ctx context.Context, msgs PushMsgs, format pushMessage
 	}
 
 	req.Header.Set("Content-Type", ApplicationJson)
+	if s.authZHeader != "" {
+		req.Header.Set("Authorization", s.authZHeader)
+	}
 
 	log.WithFields(
 		log.Fields{