Fix for Windows Event Log operator in Windows Server 2022 (open-telem…

…etry#283)

* Port fix for windows event log operator from stanza

* EvtFormatMessage uses wide charcters

* Add missing windows build tag to xml.go

operator/builtin/input/windows/bookmark.go

@@ -70,9 +70,9 @@ func (b *Bookmark) Render(buffer Buffer) (string, error) {
 	}
 
 	var bufferUsed, propertyCount uint32
-	err := evtRender(0, b.handle, EvtRenderBookmark, buffer.Size(), buffer.FirstByte(), &bufferUsed, &propertyCount)
+	err := evtRender(0, b.handle, EvtRenderBookmark, buffer.SizeBytes(), buffer.FirstByte(), &bufferUsed, &propertyCount)
 	if err == ErrorInsufficientBuffer {
-		buffer.UpdateSize(bufferUsed)
+		buffer.UpdateSizeBytes(bufferUsed)
 		return b.Render(buffer)
 	}
 

operator/builtin/input/windows/buffer.go

@@ -26,12 +26,15 @@ import (
 // defaultBufferSize is the default size of the buffer.
 const defaultBufferSize = 16384
 
+// bytesPerWChar is the number bytes in a Windows wide character.
+const bytesPerWChar = 2
+
 // Buffer is a buffer of utf-16 bytes.
 type Buffer struct {
 	buffer []byte
 }
 
-// ReadBytes will read UTF-8 bytes from the buffer.
+// ReadBytes will read UTF-8 bytes from the buffer, where offset is the number of bytes to be read
 func (b *Buffer) ReadBytes(offset uint32) ([]byte, error) {
 	utf16 := b.buffer[:offset]
 	utf8, err := unicode.UTF16(unicode.LittleEndian, unicode.UseBOM).NewDecoder().Bytes(utf16)
@@ -42,6 +45,11 @@ func (b *Buffer) ReadBytes(offset uint32) ([]byte, error) {
 	return bytes.Trim(utf8, "\u0000"), nil
 }
 
+// ReadWideChars will read UTF-8 bytes from the buffer, where offset is the number of wchars to read
+func (b *Buffer) ReadWideChars(offset uint32) ([]byte, error) {
+	return b.ReadBytes(offset * bytesPerWChar)
+}
+
 // ReadString will read a UTF-8 string from the buffer.
 func (b *Buffer) ReadString(offset uint32) (string, error) {
 	bytes, err := b.ReadBytes(offset)
@@ -51,16 +59,26 @@ func (b *Buffer) ReadString(offset uint32) (string, error) {
 	return string(bytes), nil
 }
 
-// UpdateSize will update the size of the buffer.
-func (b *Buffer) UpdateSize(size uint32) {
+// UpdateSizeBytes will update the size of the buffer to fit size bytes.
+func (b *Buffer) UpdateSizeBytes(size uint32) {
 	b.buffer = make([]byte, size)
 }
 
-// Size will return the size of the buffer.
-func (b *Buffer) Size() uint32 {
+// UpdateSizeWide will update the size of the buffer to fit size wchars.
+func (b *Buffer) UpdateSizeWide(size uint32) {
+	b.buffer = make([]byte, bytesPerWChar*size)
+}
+
+// SizeBytes will return the size of the buffer as number of bytes.
+func (b *Buffer) SizeBytes() uint32 {
 	return uint32(len(b.buffer))
 }
 
+// SizeWide returns the size of the buffer as number of wchars
+func (b *Buffer) SizeWide() uint32 {
+	return uint32(len(b.buffer) / bytesPerWChar)
+}
+
 // FirstByte will return a pointer to the first byte.
 func (b *Buffer) FirstByte() *byte {
 	return &b.buffer[0]

operator/builtin/input/windows/buffer_test.go

@@ -36,6 +36,19 @@ func TestBufferReadBytes(t *testing.T) {
 	require.Equal(t, utf8, bytes)
 }
 
+func TestBufferReadWideBytes(t *testing.T) {
+	buffer := NewBuffer()
+	utf8 := []byte("test")
+	utf16, _ := unicode.UTF16(unicode.LittleEndian, unicode.UseBOM).NewEncoder().Bytes(utf8)
+	for i, b := range utf16 {
+		buffer.buffer[i] = b
+	}
+	offset := uint32(len(utf16) / 2)
+	bytes, err := buffer.ReadWideChars(offset)
+	require.NoError(t, err)
+	require.Equal(t, utf8, bytes)
+}
+
 func TestBufferReadString(t *testing.T) {
 	buffer := NewBuffer()
 	utf8 := []byte("test")
@@ -51,13 +64,24 @@ func TestBufferReadString(t *testing.T) {
 
 func TestBufferUpdateSize(t *testing.T) {
 	buffer := NewBuffer()
-	buffer.UpdateSize(1)
+	buffer.UpdateSizeBytes(1)
 	require.Equal(t, 1, len(buffer.buffer))
 }
 
+func TestBufferUpdateSizeWide(t *testing.T) {
+	buffer := NewBuffer()
+	buffer.UpdateSizeWide(1)
+	require.Equal(t, 2, len(buffer.buffer))
+}
+
 func TestBufferSize(t *testing.T) {
 	buffer := NewBuffer()
-	require.Equal(t, uint32(defaultBufferSize), buffer.Size())
+	require.Equal(t, uint32(defaultBufferSize), buffer.SizeBytes())
+}
+
+func TestBufferSizeWide(t *testing.T) {
+	buffer := NewBuffer()
+	require.Equal(t, uint32(defaultBufferSize/2), buffer.SizeWide())
 }
 
 func TestBufferFirstByte(t *testing.T) {

operator/builtin/input/windows/event.go

@@ -32,9 +32,9 @@ func (e *Event) RenderSimple(buffer Buffer) (EventXML, error) {
 	}
 
 	var bufferUsed, propertyCount uint32
-	err := evtRender(0, e.handle, EvtRenderEventXML, buffer.Size(), buffer.FirstByte(), &bufferUsed, &propertyCount)
+	err := evtRender(0, e.handle, EvtRenderEventXML, buffer.SizeBytes(), buffer.FirstByte(), &bufferUsed, &propertyCount)
 	if err == ErrorInsufficientBuffer {
-		buffer.UpdateSize(bufferUsed)
+		buffer.UpdateSizeBytes(bufferUsed)
 		return e.RenderSimple(buffer)
 	}
 
@@ -57,17 +57,17 @@ func (e *Event) RenderFormatted(buffer Buffer, publisher Publisher) (EventXML, e
 	}
 
 	var bufferUsed uint32
-	err := evtFormatMessage(publisher.handle, e.handle, 0, 0, 0, EvtFormatMessageXML, buffer.Size(), buffer.FirstByte(), &bufferUsed)
+	err := evtFormatMessage(publisher.handle, e.handle, 0, 0, 0, EvtFormatMessageXML, buffer.SizeWide(), buffer.FirstByte(), &bufferUsed)
 	if err == ErrorInsufficientBuffer {
-		buffer.UpdateSize(bufferUsed)
+		buffer.UpdateSizeWide(bufferUsed)
 		return e.RenderFormatted(buffer, publisher)
 	}
 
 	if err != nil {
 		return EventXML{}, fmt.Errorf("syscall to 'EvtFormatMessage' failed: %s", err)
 	}
 
-	bytes, err := buffer.ReadBytes(bufferUsed)
+	bytes, err := buffer.ReadWideChars(bufferUsed)
 	if err != nil {
 		return EventXML{}, fmt.Errorf("failed to read bytes from buffer: %s", err)
 	}

operator/builtin/input/windows/operator.go

@@ -106,7 +106,8 @@ func (e *EventLogInput) Start(persister operator.Persister) error {
 	e.bookmark = NewBookmark()
 	offsetXML, err := e.getBookmarkOffset(ctx)
 	if err != nil {
-		return fmt.Errorf("failed to retrieve bookmark offset: %s", err)
+		e.Errorf("Failed to open bookmark, continuing without previous bookmark: %s", err)
+		e.persister.Delete(ctx, e.channel)
 	}
 
 	if offsetXML != "" {

operator/builtin/input/windows/xml.go

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+// +build windows
+
 package windows
 
 import (
@@ -104,7 +106,7 @@ func (e *EventXML) parseMessage() (string, map[string]interface{}) {
 func unmarshalEventXML(bytes []byte) (EventXML, error) {
 	var eventXML EventXML
 	if err := xml.Unmarshal(bytes, &eventXML); err != nil {
-		return EventXML{}, fmt.Errorf("failed to unmarshal xml bytes into event: %s", err)
+		return EventXML{}, fmt.Errorf("failed to unmarshal xml bytes into event: %w (%s)", err, string(bytes))
 	}
 	return eventXML, nil
 }

operator/builtin/input/windows/xml_test.go

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+// +build windows
+
 package windows
 
 import (