Add more detailed errors and tests in the windows/service metricset

CHANGELOG.next.asciidoc

@@ -330,6 +330,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add test for documented fields check for metricsets without a http input. {issue}17315[17315] {pull}17334[17334]
 - Add final tests and move label to GA for the azure module in metricbeat. {pull}17319[17319]
 - Reference kubernetes manifests mount data directory from the host when running metricbeat as daemonset, so data persist between executions in the same node. {pull}17429[17429]
+- Add more detailed error messages, system tests and small refactoring to the service metricset in windows. {pull}17725[17725]
 - Stack Monitoring modules now auto-configure required metricsets when `xpack.enabled: true` is set. {issue}16471[[16471] {pull}17609[17609]
 
 *Packetbeat*

libbeat/common/bytes.go

@@ -20,8 +20,22 @@ package common
 import (
 	"bytes"
 	"crypto/rand"
+	"encoding/binary"
 	"errors"
 	"fmt"
+	"io"
+	"unicode/utf16"
+	"unicode/utf8"
+)
+
+const (
+	// 0xd800-0xdc00 encodes the high 10 bits of a pair.
+	// 0xdc00-0xe000 encodes the low 10 bits of a pair.
+	// the value is those 20 bits plus 0x10000.
+	surr1           = 0xd800
+	surr2           = 0xdc00
+	surr3           = 0xe000
+	replacementChar = '\uFFFD' // Unicode replacement character
 )
 
 // Byte order utilities
@@ -76,3 +90,46 @@ func RandomBytes(length int) ([]byte, error) {
 
 	return r, nil
 }
+
+func UTF16ToUTF8Bytes(in []byte, out io.Writer) error {
+	if len(in)%2 != 0 {
+		return fmt.Errorf("input buffer must have an even length (length=%d)", len(in))
+	}
+
+	var runeBuf [4]byte
+	var v1, v2 uint16
+	for i := 0; i < len(in); i += 2 {
+		v1 = uint16(in[i]) | uint16(in[i+1])<<8
+		// Stop at null-terminator.
+		if v1 == 0 {
+			return nil
+		}
+
+		switch {
+		case v1 < surr1, surr3 <= v1:
+			n := utf8.EncodeRune(runeBuf[:], rune(v1))
+			out.Write(runeBuf[:n])
+		case surr1 <= v1 && v1 < surr2 && len(in) > i+2:
+			v2 = uint16(in[i+2]) | uint16(in[i+3])<<8
+			if surr2 <= v2 && v2 < surr3 {
+				// valid surrogate sequence
+				r := utf16.DecodeRune(rune(v1), rune(v2))
+				n := utf8.EncodeRune(runeBuf[:], r)
+				out.Write(runeBuf[:n])
+			}
+			i += 2
+		default:
+			// invalid surrogate sequence
+			n := utf8.EncodeRune(runeBuf[:], replacementChar)
+			out.Write(runeBuf[:n])
+		}
+	}
+	return nil
+}
+
+func StringToUTF16Bytes(in string) []byte {
+	var u16 []uint16 = utf16.Encode([]rune(in))
+	buf := &bytes.Buffer{}
+	binary.Write(buf, binary.LittleEndian, u16)
+	return buf.Bytes()
+}

libbeat/common/bytes_test.go

@@ -21,8 +21,10 @@ package common
 
 import (
 	"bytes"
+	"encoding/binary"
 	"errors"
 	"testing"
+	"unicode/utf16"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -256,3 +258,38 @@ func TestRandomBytes(t *testing.T) {
 	// unlikely to get 2 times the same results
 	assert.False(t, bytes.Equal(v1, v2))
 }
+
+func TestUTF16ToUTF8(t *testing.T) {
+	input := "abc白鵬翔\u145A6"
+	buf := &bytes.Buffer{}
+	binary.Write(buf, binary.LittleEndian, utf16.Encode([]rune(input)))
+	outputBuf := &bytes.Buffer{}
+	err := UTF16ToUTF8Bytes(buf.Bytes(), outputBuf)
+	assert.NoError(t, err)
+	assert.Equal(t, []byte(input), outputBuf.Bytes())
+}
+
+func TestUTF16BytesToStringTrimNullTerm(t *testing.T) {
+	input := "abc"
+	utf16Bytes := append(StringToUTF16Bytes(input), []byte{0, 0, 0, 0, 0, 0}...)
+
+	outputBuf := &bytes.Buffer{}
+	err := UTF16ToUTF8Bytes(utf16Bytes, outputBuf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	b := outputBuf.Bytes()
+	assert.Len(t, b, 3)
+	assert.Equal(t, input, string(b))
+}
+
+func BenchmarkUTF16ToUTF8(b *testing.B) {
+	utf16Bytes := StringToUTF16Bytes("A logon was attempted using explicit credentials.")
+	outputBuf := &bytes.Buffer{}
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		UTF16ToUTF8Bytes(utf16Bytes, outputBuf)
+		outputBuf.Reset()
+	}
+}

metricbeat/module/windows/service/reader.go

@@ -0,0 +1,197 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// +build windows
+
+package service
+
+import (
+	"crypto/sha256"
+	"encoding/base64"
+	"strconv"
+	"syscall"
+
+	"github.com/pkg/errors"
+	"golang.org/x/sys/windows/registry"
+
+	"github.com/elastic/beats/v7/libbeat/common"
+)
+
+var (
+	// errorNames is mapping of errno values to names.
+	// https://msdn.microsoft.com/en-us/library/windows/desktop/ms681383(v=vs.85).aspx
+	errorNames = map[uint32]string{
+		1077: "ERROR_SERVICE_NEVER_STARTED",
+	}
+	InvalidDatabaseHandle = ^Handle(0)
+)
+
+type Handle uintptr
+
+type Reader struct {
+	handle            Handle
+	state             ServiceEnumState
+	guid              string            // Host's MachineGuid value (a unique ID for the host).
+	ids               map[string]string // Cache of service IDs.
+	protectedServices map[string]struct{}
+}
+
+func NewReader() (*Reader, error) {
+	handle, err := openSCManager("", "", ScManagerEnumerateService|ScManagerConnect)
+	if err != nil {
+		return nil, errors.Wrap(err, "initialization failed")
+	}
+
+	guid, err := getMachineGUID()
+	if err != nil {
+		return nil, err
+	}
+
+	r := &Reader{
+		handle:            handle,
+		state:             ServiceStateAll,
+		guid:              guid,
+		ids:               map[string]string{},
+		protectedServices: map[string]struct{}{},
+	}
+
+	return r, nil
+}
+
+func (reader *Reader) Read() ([]common.MapStr, error) {
+	services, err := GetServiceStates(reader.handle, reader.state, reader.protectedServices)
+	if err != nil {
+		return nil, err
+	}
+
+	result := make([]common.MapStr, 0, len(services))
+
+	for _, service := range services {
+		ev := common.MapStr{
+			"id":           reader.getServiceID(service.ServiceName),
+			"display_name": service.DisplayName,
+			"name":         service.ServiceName,
+			"state":        service.CurrentState,
+			"start_type":   service.StartType.String(),
+			"start_name":   service.ServiceStartName,
+			"path_name":    service.BinaryPathName,
+		}
+
+		if service.CurrentState == "Stopped" {
+			ev.Put("exit_code", getErrorCode(service.ExitCode))
+		}
+
+		if service.PID > 0 {
+			ev.Put("pid", service.PID)
+		}
+
+		if service.Uptime > 0 {
+			if _, err = ev.Put("uptime.ms", service.Uptime); err != nil {
+				return nil, err
+			}
+		}
+
+		result = append(result, ev)
+	}
+
+	return result, nil
+}
+
+func (reader *Reader) Close() error {
+	return closeHandle(reader.handle)
+}
+
+func openSCManager(machineName string, databaseName string, desiredAccess ServiceSCMAccessRight) (Handle, error) {
+	var machineNamePtr *uint16
+	if machineName != "" {
+		var err error
+		machineNamePtr, err = syscall.UTF16PtrFromString(machineName)
+		if err != nil {
+			return InvalidDatabaseHandle, err
+		}
+	}
+
+	var databaseNamePtr *uint16
+	if databaseName != "" {
+		var err error
+		databaseNamePtr, err = syscall.UTF16PtrFromString(databaseName)
+		if err != nil {
+			return InvalidDatabaseHandle, err
+		}
+	}
+
+	handle, err := _OpenSCManager(machineNamePtr, databaseNamePtr, desiredAccess)
+	if err != nil {
+		return InvalidDatabaseHandle, ServiceErrno(err.(syscall.Errno))
+	}
+
+	return handle, nil
+}
+
+// getMachineGUID returns the machine's GUID value which is unique to a Windows
+// installation.
+func getMachineGUID() (string, error) {
+	const key = registry.LOCAL_MACHINE
+	const path = `SOFTWARE\Microsoft\Cryptography`
+	const name = "MachineGuid"
+
+	k, err := registry.OpenKey(key, path, registry.READ|registry.WOW64_64KEY)
+	if err != nil {
+		return "", errors.Wrapf(err, `failed to open HKLM\%v`, path)
+	}
+
+	guid, _, err := k.GetStringValue(name)
+	if err != nil {
+		return "", errors.Wrapf(err, `failed to get value of HKLM\%v\%v`, path, name)
+	}
+
+	return guid, nil
+}
+
+// getServiceID returns a unique ID for the service that is derived from the
+// machine's GUID and the service's name.
+func (reader *Reader) getServiceID(name string) string {
+	// hash returns a base64 encoded sha256 hash that is truncated to 10 chars.
+	hash := func(v string) string {
+		sum := sha256.Sum256([]byte(v))
+		base64Hash := base64.RawURLEncoding.EncodeToString(sum[:])
+		return base64Hash[:10]
+	}
+
+	id, found := reader.ids[name]
+	if !found {
+		id = hash(reader.guid + name)
+		reader.ids[name] = id
+	}
+
+	return id
+}
+
+func getErrorCode(errno uint32) string {
+	name, found := errorNames[errno]
+	if found {
+		return name
+	}
+	return strconv.Itoa(int(errno))
+}
+
+func closeHandle(handle Handle) error {
+	if err := _CloseServiceHandle(uintptr(handle)); err != nil {
+		return ServiceErrno(err.(syscall.Errno))
+	}
+	return nil
+}

metricbeat/module/windows/service/reader_test.go

@@ -0,0 +1,64 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// +build windows
+
+package service
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewReader(t *testing.T) {
+	reader, err := NewReader()
+	assert.NoError(t, err)
+	assert.NotNil(t, reader)
+	defer reader.Close()
+	assert.NotNil(t, reader.handle)
+}
+
+func TestOpenSCManager(t *testing.T) {
+	handle, err := openSCManager("invalidMachine", "", ScManagerEnumerateService|ScManagerConnect)
+	assert.Error(t, err)
+	assert.Equal(t, handle, InvalidDatabaseHandle)
+
+	handle, err = openSCManager("", "invalidDbName", ScManagerEnumerateService|ScManagerConnect)
+	assert.Error(t, err)
+	assert.Equal(t, handle, InvalidDatabaseHandle)
+
+	handle, err = openSCManager("", "", ScManagerEnumerateService|ScManagerConnect)
+	assert.NoError(t, err)
+	assert.NotEqual(t, handle, InvalidDatabaseHandle)
+	closeHandle(handle)
+}
+
+func TestGetMachineGUID(t *testing.T) {
+	guid, err := getMachineGUID()
+	assert.NoError(t, err)
+	assert.NotNil(t, guid)
+}
+
+func TestRead(t *testing.T) {
+	reader, err := NewReader()
+	assert.NoError(t, err)
+	result, err := reader.Read()
+	assert.NoError(t, err)
+	assert.True(t, len(result) > 0)
+	reader.Close()
+}