Merge branch 'master' into mqtt_topic_template

CHANGELOG.md

@@ -4,15 +4,22 @@
 
 ### Features
 
-- [#2154](https://github.com/influxdata/kapacitor/pull/2154): Add ability to skip ssl verification with an alert post node. Thanks @itsHabib!
 - [#2202](https://github.com/influxdata/kapacitor/pull/2202): Add templating for MQTT topics.
 
+## v1.5.3 [2019-06-18]
+
+### Features
+
+- [#2154](https://github.com/influxdata/kapacitor/pull/2154): Add ability to skip ssl verification with an alert post node. Thanks @itsHabib!
+- [#2193](https://github.com/influxdata/kapacitor/issues/2193): Add TLS configuration options.
+
 ### Bugfixes
 
 - [#2167](https://github.com/influxdata/kapacitor/pull/2167): Use default transport consistently.
 - [#2144](https://github.com/influxdata/kapacitor/issues/2144): Fix deadlock in barrier node when delete is used.
 - [#2186](https://github.com/influxdata/kapacitor/pull/2186): Make RPM create files with correct ownership on install.
 - [#2189](https://github.com/influxdata/kapacitor/pull/2189): Delete group stats when a group is deleted
+- [#2207](https://github.com/influxdata/kapacitor/pull/2207): Avoid extra allocation when building GroupID
 
 ## v1.5.2 [2018-12-12]
 

build.py

@@ -29,12 +29,12 @@
 
 INIT_SCRIPT = "scripts/init.sh"
 SYSTEMD_SCRIPT = "scripts/kapacitor.service"
+PREINST_SCRIPT = "scripts/pre-install.sh"
 POSTINST_SCRIPT = "scripts/post-install.sh"
 POSTUNINST_SCRIPT = "scripts/post-uninstall.sh"
 LOGROTATE_CONFIG = "etc/logrotate.d/kapacitor"
 BASH_COMPLETION_SH = "usr/share/bash-completion/completions/kapacitor"
 DEFAULT_CONFIG = "etc/kapacitor/kapacitor.conf"
-PREINST_SCRIPT = None
 
 # Default AWS S3 bucket for uploads
 DEFAULT_BUCKET = "dl.influxdata.com/kapacitor/artifacts"
@@ -54,6 +54,7 @@
 fpm_common_args = "-f -s dir --log error \
  --vendor {} \
  --url {} \
+ --before-install {} \
  --after-install {} \
  --after-remove {} \
  --license {} \
@@ -66,6 +67,7 @@
  --description \"{}\"".format(
         VENDOR,
         PACKAGE_URL,
+        PREINST_SCRIPT,
         POSTINST_SCRIPT,
         POSTUNINST_SCRIPT,
         PACKAGE_LICENSE,

etc/kapacitor/kapacitor.conf

@@ -29,6 +29,24 @@ default-retention-policy = ""
   ### Use a separate private key location.
   # https-private-key = ""
 
+[tls]
+  # Determines the available set of cipher suites. See https://golang.org/pkg/crypto/tls/#pkg-constants
+  # for a list of available ciphers, which depends on the version of Go (use the query
+  # SHOW DIAGNOSTICS to see the version of Go used to build Kapacitor). If not specified, uses
+  # the default settings from Go's crypto/tls package.
+  # ciphers = [
+  #   "TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305",
+  #   "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256",
+  # ]
+
+  # Minimum version of the tls protocol that will be negotiated. If not specified, uses the
+  # default settings from Go's crypto/tls package.
+  # min-version = "tls1.2"
+
+  # Maximum version of the tls protocol that will be negotiated. If not specified, uses the
+  # default settings from Go's crypto/tls package.
+  # max-version = "tls1.2"
+
 [config-override]
   # Enable/Disable the service for overridding configuration via the HTTP API.
   enabled = true

influxql.gen.go

@@ -71,7 +71,7 @@ func floatPopulateAuxFieldsAndTags(ap *influxql.FloatPoint, fieldsAndTags []stri
 func (a *floatPointAggregator) AggregatePoint(name string, p edge.FieldsTagsTimeGetter) error {
 	ap, err := convertFloatPoint(name, p, a.field, a.isSimpleSelector, a.topBottomInfo)
 	if err != nil {
-		return nil
+		return err
 	}
 	a.aggregator.AggregateFloat(ap)
 	return nil
@@ -229,7 +229,7 @@ func integerPopulateAuxFieldsAndTags(ap *influxql.IntegerPoint, fieldsAndTags []
 func (a *integerPointAggregator) AggregatePoint(name string, p edge.FieldsTagsTimeGetter) error {
 	ap, err := convertIntegerPoint(name, p, a.field, a.isSimpleSelector, a.topBottomInfo)
 	if err != nil {
-		return nil
+		return err
 	}
 	a.aggregator.AggregateInteger(ap)
 	return nil
@@ -387,7 +387,7 @@ func stringPopulateAuxFieldsAndTags(ap *influxql.StringPoint, fieldsAndTags []st
 func (a *stringPointAggregator) AggregatePoint(name string, p edge.FieldsTagsTimeGetter) error {
 	ap, err := convertStringPoint(name, p, a.field, a.isSimpleSelector, a.topBottomInfo)
 	if err != nil {
-		return nil
+		return err
 	}
 	a.aggregator.AggregateString(ap)
 	return nil
@@ -545,7 +545,7 @@ func booleanPopulateAuxFieldsAndTags(ap *influxql.BooleanPoint, fieldsAndTags []
 func (a *booleanPointAggregator) AggregatePoint(name string, p edge.FieldsTagsTimeGetter) error {
 	ap, err := convertBooleanPoint(name, p, a.field, a.isSimpleSelector, a.topBottomInfo)
 	if err != nil {
-		return nil
+		return err
 	}
 	a.aggregator.AggregateBoolean(ap)
 	return nil

influxql.gen.go.tmpl

@@ -69,7 +69,7 @@ func {{.name}}PopulateAuxFieldsAndTags(ap *influxql.{{.Name}}Point, fieldsAndTag
 func (a *{{.name}}PointAggregator) AggregatePoint(name string, p edge.FieldsTagsTimeGetter) error {
 	ap, err := convert{{.Name}}Point(name, p, a.field, a.isSimpleSelector, a.topBottomInfo)
 	if err != nil {
-		return nil
+		return err
 	}
 	a.aggregator.Aggregate{{.Name}}(ap)
 	return nil

influxql.go

@@ -114,10 +114,10 @@ func (g *influxqlGroup) BatchPoint(bp edge.BatchPointMessage) (edge.Message, err
 			return nil, nil
 		}
 	}
-	g.batchSize++
 	if err := g.rc.AggregatePoint(g.begin.Name(), bp); err != nil {
 		g.n.diag.Error("failed to aggregate point in batch", err)
 	}
+	g.batchSize++
 	return nil, nil
 }
 

integrations/helpers_test.go

@@ -1,6 +1,7 @@
 package integrations
 
 import (
+	"crypto/tls"
 	"errors"
 	"fmt"
 	"net/http"
@@ -23,7 +24,7 @@ func newHTTPDService() *httpd.Service {
 	config := httpd.NewConfig()
 	config.BindAddress = ":0" // Choose port dynamically
 	config.LogEnabled = testing.Verbose()
-	httpService := httpd.NewService(config, "localhost", diagService.NewHTTPDHandler())
+	httpService := httpd.NewService(config, "localhost", new(tls.Config), diagService.NewHTTPDHandler())
 	err := httpService.Open()
 	if err != nil {
 		panic(err)

models/point.go

@@ -1,8 +1,8 @@
 package models
 
 import (
-	"bytes"
 	"sort"
+	"strings"
 )
 
 type GroupID string
@@ -86,7 +86,7 @@ func ToGroupID(name string, tags map[string]string, dims Dimensions) GroupID {
 		}
 		return NilGroup
 	}
-	var buf bytes.Buffer
+	var buf strings.Builder
 	if dims.ByName {
 		buf.WriteString(name)
 		// Add delimiter that is not allowed in name.
@@ -101,5 +101,5 @@ func ToGroupID(name string, tags map[string]string, dims Dimensions) GroupID {
 		buf.WriteString(tags[d])
 
 	}
-	return GroupID(buf.Bytes())
+	return GroupID(buf.String())
 }

scripts/post-install.sh

@@ -26,9 +26,6 @@ function enable_chkconfig {
     chkconfig --add kapacitor
 }
 
-if ! id kapacitor >/dev/null 2>&1; then
-    useradd --system -U -M kapacitor -s /bin/false -d $DATA_DIR
-fi
 chmod a+rX $BIN_DIR/kapacitor*
 
 test -f /etc/default/kapacitor || touch /etc/default/kapacitor
@@ -44,33 +41,37 @@ if [[ -f /etc/redhat-release ]]; then
         install_init
         # Do not enable service
     fi
-else
-    mkdir -p $LOG_DIR
+elif [[ -f /etc/debian_version ]]; then
+    # Debian/Ubuntu logic
+
+    # Ownership for RH-based platforms is set in build.py via the `rmp-attr` option.
+    # We perform ownership change only for Debian-based systems.
+    # Moving these lines out of this if statement would make `rmp -V` fail after installation.
+    test -d $LOG_DIR || mkdir -p $LOG_DIR
+    test -d $DATA_DIR || mkdir -p $DATA_DIR
     chown -R -L kapacitor:kapacitor $LOG_DIR
-    mkdir -p $DATA_DIR
     chown -R -L kapacitor:kapacitor $DATA_DIR
+    chmod 755 $LOG_DIR
+    chmod 755 $DATA_DIR
 
-    if [[ -f /etc/debian_version ]]; then
-        # Debian/Ubuntu logic
-        if [[ "$(readlink /proc/1/exe)" == */systemd ]]; then
-            install_systemd
-            enable_systemd
+    if [[ "$(readlink /proc/1/exe)" == */systemd ]]; then
+        install_systemd
+        enable_systemd
+    else
+        # Assuming SysV
+        install_init
+        # Run update-rc.d or fallback to chkconfig if not available
+        if which update-rc.d &>/dev/null; then
+            enable_update_rcd
         else
-            # Assuming SysV
-            install_init
-            # Run update-rc.d or fallback to chkconfig if not available
-            if which update-rc.d &>/dev/null; then
-                enable_update_rcd
-            else
-                enable_chkconfig
-            fi
-        fi
-    elif [[ -f /etc/os-release ]]; then
-        source /etc/os-release
-        if [[ $ID = "amzn" ]]; then
-            # Amazon Linux logic
-            install_init
-            # Do not enable service
+            enable_chkconfig
         fi
     fi
+elif [[ -f /etc/os-release ]]; then
+    source /etc/os-release
+    if [[ $ID = "amzn" ]]; then
+        # Amazon Linux logic
+        install_init
+        # Do not enable service
+    fi
 fi

scripts/pre-install.sh

@@ -0,0 +1,8 @@
+#!/bin/bash
+
+DATA_DIR=/var/lib/kapacitor
+
+# create user
+if ! id kapacitor >/dev/null 2>&1; then
+    useradd --system -U -M kapacitor -s /bin/false -d $DATA_DIR
+fi

server/config.go

@@ -57,6 +57,7 @@ import (
 	"github.com/influxdata/kapacitor/services/udf"
 	"github.com/influxdata/kapacitor/services/udp"
 	"github.com/influxdata/kapacitor/services/victorops"
+	"github.com/influxdata/kapacitor/tlsconfig"
 	"github.com/pkg/errors"
 
 	"github.com/influxdata/influxdb/services/collectd"
@@ -75,6 +76,7 @@ type Config struct {
 	InfluxDB       []influxdb.Config `toml:"influxdb" override:"influxdb,element-key=name"`
 	Logging        diagnostic.Config `toml:"logging"`
 	ConfigOverride config.Config     `toml:"config-override"`
+	TLS            tlsconfig.Config  `toml:"tls"`
 
 	// Input services
 	Graphite []graphite.Config `toml:"graphite"`
@@ -147,6 +149,7 @@ func NewConfig() *Config {
 	c.InfluxDB = []influxdb.Config{influxdb.NewConfig()}
 	c.Logging = diagnostic.NewConfig()
 	c.ConfigOverride = config.NewConfig()
+	c.TLS = tlsconfig.NewConfig()
 
 	c.Collectd = collectd.NewConfig()
 	c.OpenTSDB = opentsdb.NewConfig()
@@ -222,6 +225,9 @@ func (c *Config) Validate() error {
 	if err := c.Task.Validate(); err != nil {
 		return errors.Wrap(err, "task")
 	}
+	if err := c.TLS.Validate(); err != nil {
+		return errors.Wrap(err, "tls")
+	}
 	if err := c.Load.Validate(); err != nil {
 		return err
 	}

server/server.go

@@ -2,6 +2,7 @@
 package server
 
 import (
+	"crypto/tls"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -98,7 +99,8 @@ type Server struct {
 	dataDir  string
 	hostname string
 
-	config *Config
+	config    *Config
+	tlsConfig *tls.Config
 
 	err chan error
 
@@ -158,9 +160,18 @@ func New(c *Config, buildInfo BuildInfo, diagService *diagnostic.Service) (*Serv
 	if err != nil {
 		return nil, fmt.Errorf("invalid configuration: %s. To generate a valid configuration file run `kapacitord config > kapacitor.generated.conf`.", err)
 	}
+	// Setup base TLS config used for the Kapacitor API
+	tlsConfig, err := c.TLS.Parse()
+	if err != nil {
+		return nil, errors.Wrap(err, "tls configuration")
+	}
+	if tlsConfig == nil {
+		tlsConfig = new(tls.Config)
+	}
 	d := diagService.NewServerHandler()
 	s := &Server{
 		config:           c,
+		tlsConfig:        tlsConfig,
 		BuildInfo:        buildInfo,
 		dataDir:          c.DataDir,
 		hostname:         c.Hostname,
@@ -448,7 +459,7 @@ func (s *Server) appendInfluxDBService() error {
 
 func (s *Server) initHTTPDService() {
 	d := s.DiagService.NewHTTPDHandler()
-	srv := httpd.NewService(s.config.HTTP, s.hostname, d)
+	srv := httpd.NewService(s.config.HTTP, s.hostname, s.tlsConfig, d)
 
 	srv.LocalHandler.PointsWriter = s.TaskMaster
 	srv.Handler.PointsWriter = s.TaskMaster

services/httpd/service.go

@@ -58,12 +58,13 @@ type Diagnostic interface {
 }
 
 type Service struct {
-	ln    net.Listener
-	addr  string
-	https bool
-	cert  string
-	key   string
-	err   chan error
+	ln        net.Listener
+	addr      string
+	https     bool
+	cert      string
+	tlsConfig *tls.Config
+	key       string
+	err       chan error
 
 	externalURL string
 
@@ -87,7 +88,7 @@ type Service struct {
 	httpServerErrorLogger *log.Logger
 }
 
-func NewService(c Config, hostname string, d Diagnostic) *Service {
+func NewService(c Config, hostname string, t *tls.Config, d Diagnostic) *Service {
 	statMap := &expvar.Map{}
 	statMap.Init()
 
@@ -108,6 +109,7 @@ func NewService(c Config, hostname string, d Diagnostic) *Service {
 		key:             c.HTTPSPrivateKey,
 		externalURL:     u.String(),
 		err:             make(chan error, 1),
+		tlsConfig:       t,
 		shutdownTimeout: time.Duration(c.ShutdownTimeout),
 		Handler: NewHandler(
 			c.AuthEnabled,
@@ -153,9 +155,9 @@ func (s *Service) Open() error {
 			return err
 		}
 
-		listener, err := tls.Listen("tcp", s.addr, &tls.Config{
-			Certificates: []tls.Certificate{cert},
-		})
+		tlsConfig := s.tlsConfig.Clone()
+		tlsConfig.Certificates = []tls.Certificate{cert}
+		listener, err := tls.Listen("tcp", s.addr, tlsConfig)
 		if err != nil {
 			return err
 		}