Scott Weinstein on .Net

The PowerShell team has a short post on using V2 cmdlets to Monitor performance counters.

Building on that, and the prior work with the PoShAdapter, here’s a sample with Coral8 and PowerShell to alert you via SMS when the rolling 30 seconds average CPU hits a threshold.

create VARIABLE float CPUThreashold  = 70; 
create OUTPUT STREAM fakeout SCHEMA (x string);
CREATE SCHEMA CPUSchema (cpu FLOAT);
create INPUT STREAM inCPU SCHEMA CPUSchema;
create OUTPUT STREAM highCPU SCHEMA CPUSchema;
create local STREAM avgCPU SCHEMA CPUSchema;

ATTACH OUTPUT ADAPTER smsAdapter TYPE PoShAdapter TO STREAM highCPU
PROPERTIES
    BEGINBLOCK  = [[ 
    Add-Type -AssemblyName System.Security   
    Add-Type -Path "C:\Program Files\Coral8\Server\bin\GmailHelper.dll" 
    $pass = ".."
    ]],
PROCESSBlock     = [[
 foreach ($cpuSpike in $input)
 {
  $m = "CPU spike at {0:0.00}" -f $cpuSpike['cpu']
  [RC.Gmail.GmailMessage]::SendFromGmail("user", $pass,"6465555555@tmomail.net", 
   "Via Powershell/Coral8",$m)
 }
]];

ATTACH OUTPUT ADAPTER CPUAdapter TYPE PoShAdapter TO STREAM fakeout
PROPERTIES
    RESULTSSTREAM = "ccl://localhost:6789/Stream/Default/MonitorCPU/inCPU",
    INPUTBLOCK    = [[
Get-Counter '\Processor(*)\% Processor Time'    |
    select -expand CounterSamples               |
        ? { $_.InstanceName -eq "_total" }      | 
            % { ,,($_.CookedValue,0) } # 
]];

INSERT into avgCPU
select avg(cpu)  
from inCPU 
keep 30 SECONDS 
OUTPUT EVERY 30 SECONDS;

INSERT into highCPU 
select cpu 
from avgCPU 
WHERE cpu > CPUThreashold;

Yields:

*I lowered the threshold for testing
** Yes, I should check my voice mail

The system I helped build at my previous job gets a nice write-up in Wall Street & Technology:

Citi Taps CEP for Analyzing Equity Data

I’m sure Marc, Hanno and the rest of the team are pleased.

Congrats guys!

I’ve updated the PowerShell Coral8 adapter so it can be used to receive input. Rather than demonstrate input from a database or an RSS stream, both of which can are supported by the native Coral8 adapters, I have a demo with WMI events. In this case allowing us to monitor process creation across a network.

Using the same general template as before we have the following two script blocks

BeginBlock

$computers = &{$args} localhost ccs01 ccs02
$jobs = $computers | % `
 {
   Register-WmiEvent -Class Win32_ProcessStartTrace -ComputerName $_ ` 
   -Action  { `
            $res = @{}
            $res['ProcessName'] = $args[1].NewEvent.ProcessName
            $res['ComputerName'] = $args[0].scope.Path.Server
            return $res
        }
    }

InputBlock

while ($true) 
{
    Start-Sleep -Milliseconds 100
    foreach ($p in (Receive-Job $jobs))
    {
        # Input adapter doesn't support hashtables yet
        ,,($p['ProcessName'],$p['ComputerName']) 
    }
}

And the payoff

In the first post on integrating PowerShell and Coral8 I showed how to create a message sink. In C#ish pseudocode we did the following:

OutputStream<T>  =>  Action<T>

where the Action<T> was a PowerShell block to send a message via GMail.

Today the goal is to do a transform, something along the following

OutputStream<T>  =>  Func<T,T2> => InputStream<T2>

In this example, T will be a pair of stock symbols, T2 will be a pair stock symbols along with the correlation of their log normal closing prices for the past 10 days.

let’s get started.

CREATE SCHEMA           StockPairSchema( SecA STRING,SecB STRING );
CREATE SCHEMA           StockPairCorrelationSchema INHERITS from StockPairSchema (CORR  FLOAT) ;
CREATE OUTPUT   STREAM  StockPairs          SCHEMA StockPairSchema;
CREATE INPUT    STREAM  StockCorrelations   SCHEMA StockPairCorrelationSchema;
ATTACH OUTPUT ADAPTER PairstoCorrelationsFunc TYPE PoShAdapter TO STREAM StockPairs
PROPERTIES
RESULTSSTREAM = "ccl://localhost:6789/Stream/Default/TestC8/StockCorrelations",

The process block is very simple:

foreach ($t in $input)
{
    $a = Get-LogReturns $t["SecA"]
    $b = Get-LogReturns $t["SecB"]
    $c = [Demo.Stats]::Correlate($a, $b)
    ,,($t["SecA"],$t["SecB"],$c) #// double commas so the values are not flattened
}

The block path defines the Get-LogReturns and Correlate function, so naturally it’s a bit longer.
Get-LogReturns:

$wc = New-Object Net.WebClient    
function Get-LogReturns ($sec)
{
    $qry = "http://ichart.finance.yahoo.com/table.csv?s=$sec"
    $secAData = ConvertFrom-Csv $wc.DownloadString($qry) | select  -First 10 | % { $_.'Adj Close' }
    for ($i=0; $i -lt $secAData.Count-1; $i++) { [Math]::Log( $secAData[$i] / $secAData[$i+1] )  }
}

[Demo.Stats]::Correlate

$csCode = @"
using System;
using System.Collections.Generic;
using System.Linq;
namespace Demo 
{
    public static class Stats
    {
        private static IEnumerable<TResult> Zip<TFirst, TSecond, TResult>(IList<TFirst> first, 
				IList<TSecond> second, 
				Func<TFirst, TSecond, TResult> func)
        {
            for (int ii = 0; ii < Math.Min(first.Count(),second.Count()); ii++)
                yield return func(first[ii],second[ii]);
        }
        public static double Correlate(object[] s1,object[] s2)
        {
            return Correlate(s1.Cast<double>(), s2.Cast<double>());
        }
        public static double Correlate(IEnumerable<double> s1,IEnumerable<double> s2)
        {
            var sum1 = s1.Sum();
            var sum2 = s2.Sum();
            var sumSq1 = s1.Select(v=> v*v).Sum();
            var sumSq2 = s2.Select(v=> v*v).Sum();
            var pSum = Zip(s1.ToList(),s2.ToList(), ( a, b) => a*b).Sum();
            var len = s1.Count();
            var num = pSum - ((sum1 * sum2) / len);
            var denom = Math.Sqrt(((sumSq1 - (sum1 * sum1) / len) * (sumSq2 - (sum2 * sum2) / len)));
            return (denom == 0.0) ? 0 : num / denom;
        }
    }
}
"@
Add-Type -TypeDefinition $csCode -Language CSharpVersion3 -PassThru 

And in case you missed it, the correlate function isn't in PowerShell at all, but rather coded up via in-line C# code, compiled at the startup of the adaptor, and running in the Coral8 server process.

For certain types, PowerShell has automatic properties, by this I mean PowerShell is able to inspect the object and then expose properties where normally you would need to use a string indexer.

Like so:

$xbooks = [xml]"<Books><Book><Title>PowerShell in Action</Title></Book></Books>" 
$xbooks.Books.Book.Title

Similar functionality is available for COM, WMI, ADO.Net, and AD types. It’s one of the core features of PowerShell, and its simplicity and elegance makes even diehard UNIX types sit up and take notice.

But if you’re writing your own cmdlet, navigation provider, or are hosting PowerShell, and you wish to provide an adapter for your own type – you will quickly find that adapters and rest of the infrastructure for automatic properties are internal classes and unavailable to you.

The best option you have is something along the following:

public PSObject CreateCustomPSObject(MyType mytype )
{
    PSObject pso = new PSObject(mytype);
    foreach (string name in MyColumnNames)     
    { 
        string getter = String.Format("$this[\"{0}\"]", name); 
        string setter = String.Format("$this[\"{0}\"] = $value", name); 
        pso.Properties.Add(new PSScriptProperty(name,                                        InvokeCommand.NewScriptBlock(getter),                                InvokeCommand.NewScriptBlock(setter))); 
    
    } 
    return pso; 
}
WriteItemObject(CreateCustomPSObject(mytype));

This will work, but the output from get-member won’t be as clean as it is for a DataRow or other adapted types. Even worse, this option isn’t available at all for hosted scenarios as there no instance of CommandInvocationIntrinsics available from a Runspace or Pipleline.

A few days ago a thought crept into my head; wouldn’t it be nice if I could have a Coral8 in-process adapter to run PowerShell code? If it worked it could potentially be a “universal” adapter, limiting the need for custom one-off adapters.

Coral8, out-of-the-box, supports two different types of adapters. In-process adapters which must be written in C (enough said?), and out-of-process adapters which can be written in the language of your choice, but then have startup sequence, configuration, and deployment work associated with them.

Using the PoShAdapter I’ve been developing, let’s take a look at the adapter as a message sink. In this case a  “sent to Gmail” adapter.

CREATE SCHEMA TSchema ( toAddr STRING , subject string, message as string); 
CREATE OUTPUT STREAM outGmail SCHEMA TSchema; 
ATTACH OUTPUT ADAPTER PoShAdapter TYPE PoShAdapter  TO STREAM outGmail  PROPERTIES          
BEGINBLOCK  = [[   
    Add-Type -AssemblyName System.Security     
    Add-Type -Path "C:\Program Files\Coral8\Server\bin\GmailHelper.dll"   
    $encPassword = "AQQbRXQr…kg=="   
    $password =[Text.Encoding]::UTF8.GetString([Security.Cryptography.ProtectedData]::UnProtect([System.Convert]::FromBase64String($encPassword),$null,"CurrentUser"))
]],
PROCESS     = [[             
    foreach ($t in $input)             
    {                  
        [RC.Gmail.GmailMessage]::SendFromGmail("from.user", $password,
                       $t["toAddr"], $t["subject"], $t[“message”])
    }
    ]];

The posh adapter, like PowerShell scripts and functions have begin, process, and end blocks.

In the beginblock we load the assemblies for System.Security and GmailHelper, as well as decrypt the gmail password.

The process block will have as its $input set to a collection of the Coral8 tuples, so it simply loops through them sends out an email for each one. Because we the Runspace is kept for the lifetime of the adapter, context from the beginblock is available in the process block.

In later posts I’ll explore using the PoShAdapter to transform data (as in a select or map operator), and look into the possibility of using the adapter to listen for WMI or other system events.