Allow globs in FPM unix socket paths

plugins/inputs/phpfpm/README.md

@@ -19,6 +19,8 @@ Get phpfpm stats using either HTTP status page or fpm socket.
   ##       "/var/run/php5-fpm.sock"
   ##      or using a custom fpm status path:
   ##       "/var/run/php5-fpm.sock:fpm-custom-status-path"
+  ##      glob patterns are also supported:
+  ##       "/var/run/php*.sock"
   ##
   ##   - fcgi: the URL must start with fcgi:// or cgi://, and port must be present, ie:
   ##       "fcgi://10.0.0.12:9000/status"

plugins/inputs/phpfpm/phpfpm.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
+	"github.com/influxdata/telegraf/internal/globpath"
 	"github.com/influxdata/telegraf/internal/tls"
 	"github.com/influxdata/telegraf/plugins/inputs"
 )
@@ -95,7 +96,12 @@ func (g *phpfpm) Gather(acc telegraf.Accumulator) error {
 
 	var wg sync.WaitGroup
 
-	for _, serv := range g.Urls {
+	urls, err := expandUrls(g.Urls)
+	if err != nil {
+		return err
+	}
+
+	for _, serv := range urls {
 		wg.Add(1)
 		go func(serv string) {
 			defer wg.Done()
@@ -153,18 +159,10 @@ func (g *phpfpm) gatherServer(addr string, acc telegraf.Accumulator) error {
 			statusPath = "status"
 		}
 	} else {
-		socketAddr := strings.Split(addr, ":")
-		if len(socketAddr) >= 2 {
-			socketPath = socketAddr[0]
-			statusPath = socketAddr[1]
-		} else {
-			socketPath = socketAddr[0]
+		socketPath, statusPath = unixSocketPaths(addr)
+		if statusPath == "" {
 			statusPath = "status"
 		}
-
-		if _, err := os.Stat(socketPath); os.IsNotExist(err) {
-			return fmt.Errorf("Socket doesn't exist  '%s': %s", socketPath, err)
-		}
 		fcgi, err = newFcgiClient("unix", socketPath)
 	}
 
@@ -277,6 +275,75 @@ func importMetric(r io.Reader, acc telegraf.Accumulator, addr string) (poolStat,
 	return stats, nil
 }
 
+func expandUrls(urls []string) ([]string, error) {
+	addrs := make([]string, 0, len(urls))
+	for _, url := range urls {
+		if isNetworkURL(url) {
+			addrs = append(addrs, url)
+			continue
+		}
+		paths, err := globUnixSocket(url)
+		if err != nil {
+			return nil, err
+		}
+		addrs = append(addrs, paths...)
+	}
+	return addrs, nil
+}
+
+func globUnixSocket(url string) ([]string, error) {
+	pattern, status := unixSocketPaths(url)
+	glob, err := globpath.Compile(pattern)
+	if err != nil {
+		return nil, fmt.Errorf("could not compile glob %q: %v", pattern, err)
+	}
+	paths := glob.Match()
+	if len(paths) == 0 {
+		return nil, nil
+	}
+
+	// globpath.Match() returns the given argument when it's a static path,
+	// i.e., not a glob pattern. In that case, ensure it exists.
+	if len(paths) == 1 && paths[0] == pattern {
+		if _, err := os.Stat(paths[0]); err != nil {
+			if os.IsNotExist(err) {
+				return nil, fmt.Errorf("Socket doesn't exist  '%s': %s", pattern, err)
+			}
+			return nil, err
+		}
+	}
+
+	addrs := make([]string, 0, len(paths))
+
+	for _, path := range paths {
+		if status != "" {
+			status = fmt.Sprintf(":%s", status)
+		}
+		addrs = append(addrs, fmt.Sprintf("%s%s", path, status))
+	}
+
+	return addrs, nil
+}
+
+func unixSocketPaths(addr string) (string, string) {
+	var socketPath, statusPath string
+
+	socketAddr := strings.Split(addr, ":")
+	if len(socketAddr) >= 2 {
+		socketPath = socketAddr[0]
+		statusPath = socketAddr[1]
+	} else {
+		socketPath = socketAddr[0]
+		statusPath = ""
+	}
+
+	return socketPath, statusPath
+}
+
+func isNetworkURL(addr string) bool {
+	return strings.HasPrefix(addr, "http://") || strings.HasPrefix(addr, "https://") || strings.HasPrefix(addr, "fcgi://") || strings.HasPrefix(addr, "cgi://")
+}
+
 func init() {
 	inputs.Add("phpfpm", func() telegraf.Input {
 		return &phpfpm{}

plugins/inputs/phpfpm/phpfpm_test.go

@@ -148,6 +148,71 @@ func TestPhpFpmGeneratesMetrics_From_Socket(t *testing.T) {
 	acc.AssertContainsTaggedFields(t, "phpfpm", fields, tags)
 }
 
+func TestPhpFpmGeneratesMetrics_From_Multiple_Sockets_With_Glob(t *testing.T) {
+	// Create a socket in /tmp because we always have write permission and if the
+	// removing of socket fail when system restart /tmp is clear so
+	// we don't have junk files around
+	var randomNumber int64
+	binary.Read(rand.Reader, binary.LittleEndian, &randomNumber)
+	socket1 := fmt.Sprintf("/tmp/test-fpm%d.sock", randomNumber)
+	tcp1, err := net.Listen("unix", socket1)
+	if err != nil {
+		t.Fatal("Cannot initialize server on port ")
+	}
+	defer tcp1.Close()
+
+	binary.Read(rand.Reader, binary.LittleEndian, &randomNumber)
+	socket2 := fmt.Sprintf("/tmp/test-fpm%d.sock", randomNumber)
+	tcp2, err := net.Listen("unix", socket2)
+	if err != nil {
+		t.Fatal("Cannot initialize server on port ")
+	}
+	defer tcp2.Close()
+
+	s := statServer{}
+	go fcgi.Serve(tcp1, s)
+	go fcgi.Serve(tcp2, s)
+
+	r := &phpfpm{
+		Urls: []string{"/tmp/test-fpm[\\-0-9]*.sock"},
+	}
+
+	var acc1, acc2 testutil.Accumulator
+
+	err = acc1.GatherError(r.Gather)
+	require.NoError(t, err)
+
+	err = acc2.GatherError(r.Gather)
+	require.NoError(t, err)
+
+	tags1 := map[string]string{
+		"pool": "www",
+		"url":  socket1,
+	}
+
+	tags2 := map[string]string{
+		"pool": "www",
+		"url":  socket2,
+	}
+
+	fields := map[string]interface{}{
+		"start_since":          int64(1991),
+		"accepted_conn":        int64(3),
+		"listen_queue":         int64(1),
+		"max_listen_queue":     int64(0),
+		"listen_queue_len":     int64(0),
+		"idle_processes":       int64(1),
+		"active_processes":     int64(1),
+		"total_processes":      int64(2),
+		"max_active_processes": int64(1),
+		"max_children_reached": int64(2),
+		"slow_requests":        int64(1),
+	}
+
+	acc1.AssertContainsTaggedFields(t, "phpfpm", fields, tags1)
+	acc2.AssertContainsTaggedFields(t, "phpfpm", fields, tags2)
+}
+
 func TestPhpFpmGeneratesMetrics_From_Socket_Custom_Status_Path(t *testing.T) {
 	// Create a socket in /tmp because we always have write permission. If the
 	// removing of socket fail we won't have junk files around. Cuz when system