Merge pull request #979 from dcantah/hostprocess-stats

Add stats support for job containers

internal/jobcontainers/jobcontainer.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"syscall"
 	"time"
+	"unsafe"
 
 	"github.com/Microsoft/go-winio/pkg/guid"
 	"github.com/Microsoft/hcsshim/internal/cow"
@@ -20,6 +21,7 @@ import (
 	"github.com/Microsoft/hcsshim/internal/queue"
 	"github.com/Microsoft/hcsshim/internal/schema1"
 	hcsschema "github.com/Microsoft/hcsshim/internal/schema2"
+	"github.com/Microsoft/hcsshim/internal/winapi"
 	specs "github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -370,12 +372,91 @@ func (c *JobContainer) shutdown(ctx context.Context) error {
 // to adhere to the interface for containers on Windows it is partially implemented. The only
 // supported property is schema2.PTStatistics.
 func (c *JobContainer) PropertiesV2(ctx context.Context, types ...hcsschema.PropertyType) (*hcsschema.Properties, error) {
-	return nil, errors.New("`PropertiesV2` call is not implemented for job containers")
+	if len(types) == 0 {
+		return nil, errors.New("no property types supplied for PropertiesV2 call")
+	}
+	if types[0] != hcsschema.PTStatistics {
+		return nil, errors.New("PTStatistics is the only supported property type for job containers")
+	}
+
+	memInfo, err := c.job.QueryMemoryStats()
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to query for job containers memory information")
+	}
+
+	processorInfo, err := c.job.QueryProcessorStats()
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to query for job containers processor information")
+	}
+
+	storageInfo, err := c.job.QueryStorageStats()
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to query for job containers storage information")
+	}
+
+	var privateWorkingSet uint64
+	err = forEachProcessInfo(c.job, func(procInfo *winapi.SYSTEM_PROCESS_INFORMATION) {
+		privateWorkingSet += uint64(procInfo.WorkingSetPrivateSize)
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get private working set for container")
+	}
+
+	return &hcsschema.Properties{
+		Statistics: &hcsschema.Statistics{
+			Timestamp:          time.Now(),
+			Uptime100ns:        uint64(time.Since(c.startTimestamp)) / 100,
+			ContainerStartTime: c.startTimestamp,
+			Memory: &hcsschema.MemoryStats{
+				MemoryUsageCommitBytes:            memInfo.JobMemory,
+				MemoryUsageCommitPeakBytes:        memInfo.PeakJobMemoryUsed,
+				MemoryUsagePrivateWorkingSetBytes: privateWorkingSet,
+			},
+			Processor: &hcsschema.ProcessorStats{
+				RuntimeKernel100ns: uint64(processorInfo.TotalKernelTime),
+				RuntimeUser100ns:   uint64(processorInfo.TotalUserTime),
+				TotalRuntime100ns:  uint64(processorInfo.TotalKernelTime + processorInfo.TotalUserTime),
+			},
+			Storage: &hcsschema.StorageStats{
+				ReadCountNormalized:  storageInfo.IoInfo.ReadOperationCount,
+				ReadSizeBytes:        storageInfo.IoInfo.ReadTransferCount,
+				WriteCountNormalized: storageInfo.IoInfo.WriteOperationCount,
+				WriteSizeBytes:       storageInfo.IoInfo.WriteTransferCount,
+			},
+		},
+	}, nil
 }
 
-// Properties is not implemented for job containers. This is just to satisfy the cow.Container interface.
+// Properties returns properties relating to the job container. This is an HCS construct but
+// to adhere to the interface for containers on Windows it is partially implemented. The only
+// supported property is schema1.PropertyTypeProcessList.
 func (c *JobContainer) Properties(ctx context.Context, types ...schema1.PropertyType) (*schema1.ContainerProperties, error) {
-	return nil, errors.New("`Properties` call is not implemented for job containers")
+	if len(types) == 0 {
+		return nil, errors.New("no property types supplied for Properties call")
+	}
+	if types[0] != schema1.PropertyTypeProcessList {
+		return nil, errors.New("ProcessList is the only supported property type for job containers")
+	}
+
+	var processList []schema1.ProcessListItem
+	err := forEachProcessInfo(c.job, func(procInfo *winapi.SYSTEM_PROCESS_INFORMATION) {
+		proc := schema1.ProcessListItem{
+			CreateTimestamp:              time.Unix(0, procInfo.CreateTime),
+			ProcessId:                    uint32(procInfo.UniqueProcessID),
+			ImageName:                    procInfo.ImageName.String(),
+			UserTime100ns:                uint64(procInfo.UserTime),
+			KernelTime100ns:              uint64(procInfo.KernelTime),
+			MemoryCommitBytes:            uint64(procInfo.PrivatePageCount),
+			MemoryWorkingSetPrivateBytes: uint64(procInfo.WorkingSetPrivateSize),
+			MemoryWorkingSetSharedBytes:  uint64(procInfo.WorkingSetSize) - uint64(procInfo.WorkingSetPrivateSize),
+		}
+		processList = append(processList, proc)
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get process ")
+	}
+
+	return &schema1.ContainerProperties{ProcessList: processList}, nil
 }
 
 // Terminate terminates the job object (kills every process in the job).
@@ -449,3 +530,73 @@ func (c *JobContainer) IsOCI() bool {
 func (c *JobContainer) OS() string {
 	return "windows"
 }
+
+// For every process in the job `job`, run the function `work`. This can be used to grab/filter the SYSTEM_PROCESS_INFORMATION
+// data from every process in a job.
+func forEachProcessInfo(job *jobobject.JobObject, work func(*winapi.SYSTEM_PROCESS_INFORMATION)) error {
+	procInfos, err := systemProcessInformation()
+	if err != nil {
+		return err
+	}
+
+	pids, err := job.Pids()
+	if err != nil {
+		return err
+	}
+
+	pidsMap := make(map[uint32]struct{})
+	for _, pid := range pids {
+		pidsMap[pid] = struct{}{}
+	}
+
+	for _, procInfo := range procInfos {
+		if _, ok := pidsMap[uint32(procInfo.UniqueProcessID)]; ok {
+			work(procInfo)
+		}
+	}
+	return nil
+}
+
+// Get a slice of SYSTEM_PROCESS_INFORMATION for all of the processes running on the system.
+func systemProcessInformation() ([]*winapi.SYSTEM_PROCESS_INFORMATION, error) {
+	var (
+		systemProcInfo *winapi.SYSTEM_PROCESS_INFORMATION
+		procInfos      []*winapi.SYSTEM_PROCESS_INFORMATION
+		// This happens to be the buffer size hcs uses but there's no really no hard need to keep it
+		// the same, it's just a sane default.
+		size   = uint32(1024 * 512)
+		bounds uintptr
+	)
+	for {
+		b := make([]byte, size)
+		systemProcInfo = (*winapi.SYSTEM_PROCESS_INFORMATION)(unsafe.Pointer(&b[0]))
+		status := winapi.NtQuerySystemInformation(
+			winapi.SystemProcessInformation,
+			uintptr(unsafe.Pointer(systemProcInfo)),
+			size,
+			&size,
+		)
+		if winapi.NTSuccess(status) {
+			// Cache the address of the end of our buffer so we can check we don't go past this
+			// in some odd case.
+			bounds = uintptr(unsafe.Pointer(&b[len(b)-1]))
+			break
+		} else if status != winapi.STATUS_INFO_LENGTH_MISMATCH {
+			return nil, winapi.RtlNtStatusToDosError(status)
+		}
+	}
+
+	for {
+		if uintptr(unsafe.Pointer(systemProcInfo))+uintptr(systemProcInfo.NextEntryOffset) >= bounds {
+			// The next entry is outside of the bounds of our buffer somehow, abort.
+			return nil, errors.New("system process info entry exceeds allocated buffer")
+		}
+		procInfos = append(procInfos, systemProcInfo)
+		if systemProcInfo.NextEntryOffset == 0 {
+			break
+		}
+		systemProcInfo = (*winapi.SYSTEM_PROCESS_INFORMATION)(unsafe.Pointer(uintptr(unsafe.Pointer(systemProcInfo)) + uintptr(systemProcInfo.NextEntryOffset)))
+	}
+
+	return procInfos, nil
+}

internal/jobcontainers/system_test.go

@@ -0,0 +1,12 @@
+package jobcontainers
+
+import (
+	"testing"
+)
+
+func TestSystemInfo(t *testing.T) {
+	_, err := systemProcessInformation()
+	if err != nil {
+		t.Fatal(err)
+	}
+}

internal/winapi/system.go

@@ -0,0 +1,52 @@
+package winapi
+
+import "golang.org/x/sys/windows"
+
+const SystemProcessInformation = 5
+
+const STATUS_INFO_LENGTH_MISMATCH = 0xC0000004
+
+// __kernel_entry NTSTATUS NtQuerySystemInformation(
+// 	SYSTEM_INFORMATION_CLASS SystemInformationClass,
+// 	PVOID                    SystemInformation,
+// 	ULONG                    SystemInformationLength,
+// 	PULONG                   ReturnLength
+// );
+//sys NtQuerySystemInformation(systemInfoClass int, systemInformation uintptr, systemInfoLength uint32, returnLength *uint32) (status uint32) = ntdll.NtQuerySystemInformation
+
+type SYSTEM_PROCESS_INFORMATION struct {
+	NextEntryOffset              uint32         // ULONG
+	NumberOfThreads              uint32         // ULONG
+	WorkingSetPrivateSize        int64          // LARGE_INTEGER
+	HardFaultCount               uint32         // ULONG
+	NumberOfThreadsHighWatermark uint32         // ULONG
+	CycleTime                    uint64         // ULONGLONG
+	CreateTime                   int64          // LARGE_INTEGER
+	UserTime                     int64          // LARGE_INTEGER
+	KernelTime                   int64          // LARGE_INTEGER
+	ImageName                    UnicodeString  // UNICODE_STRING
+	BasePriority                 int32          // KPRIORITY
+	UniqueProcessID              windows.Handle // HANDLE
+	InheritedFromUniqueProcessID windows.Handle // HANDLE
+	HandleCount                  uint32         // ULONG
+	SessionID                    uint32         // ULONG
+	UniqueProcessKey             *uint32        // ULONG_PTR
+	PeakVirtualSize              uintptr        // SIZE_T
+	VirtualSize                  uintptr        // SIZE_T
+	PageFaultCount               uint32         // ULONG
+	PeakWorkingSetSize           uintptr        // SIZE_T
+	WorkingSetSize               uintptr        // SIZE_T
+	QuotaPeakPagedPoolUsage      uintptr        // SIZE_T
+	QuotaPagedPoolUsage          uintptr        // SIZE_T
+	QuotaPeakNonPagedPoolUsage   uintptr        // SIZE_T
+	QuotaNonPagedPoolUsage       uintptr        // SIZE_T
+	PagefileUsage                uintptr        // SIZE_T
+	PeakPagefileUsage            uintptr        // SIZE_T
+	PrivatePageCount             uintptr        // SIZE_T
+	ReadOperationCount           int64          // LARGE_INTEGER
+	WriteOperationCount          int64          // LARGE_INTEGER
+	OtherOperationCount          int64          // LARGE_INTEGER
+	ReadTransferCount            int64          // LARGE_INTEGER
+	WriteTransferCount           int64          // LARGE_INTEGER
+	OtherTransferCount           int64          // LARGE_INTEGER
+}

internal/winapi/thread.go

@@ -1,3 +1,12 @@
 package winapi
 
+// HANDLE CreateRemoteThread(
+// 	HANDLE                 hProcess,
+// 	LPSECURITY_ATTRIBUTES  lpThreadAttributes,
+// 	SIZE_T                 dwStackSize,
+// 	LPTHREAD_START_ROUTINE lpStartAddress,
+// 	LPVOID                 lpParameter,
+// 	DWORD                  dwCreationFlags,
+// 	LPDWORD                lpThreadId
+// );
 //sys CreateRemoteThread(process windows.Handle, sa *windows.SecurityAttributes, stackSize uint32, startAddr uintptr, parameter uintptr, creationFlags uint32, threadID *uint32) (handle windows.Handle, err error) = kernel32.CreateRemoteThread

internal/winapi/winapi.go

@@ -2,4 +2,4 @@
 // be thought of as an extension to golang.org/x/sys/windows.
 package winapi
 
-//go:generate go run ..\..\mksyscall_windows.go -output zsyscall_windows.go net.go path.go thread.go iocp.go jobobject.go logon.go memory.go process.go processor.go devices.go filesystem.go errors.go
+//go:generate go run ..\..\mksyscall_windows.go -output zsyscall_windows.go system.go net.go path.go thread.go iocp.go jobobject.go logon.go memory.go process.go processor.go devices.go filesystem.go errors.go