Rest Interface to Elasticsearch query...

ccgw.go

package main                     

/*

To Get Results of a CC: [95% implemented]
curl -X GET http://localhost:12345/cc/742dd936-73f3-4bbf-850e-ff951af042e/results


To Trigger a CC: [50% implemented - does NOT trigger cc_analyzer...]
curl -X POST http://localhost:12345/cc -H 'Content-Type: application/json' \
            -d '{"fcheck":"all", "filters":"vrf=pepsi"}'                    

To Cancel a prior CC Run: [0% implemented]
curl -X DELETE http://localhost:12345/cc/742dd936-73f3-4bbf-850e-ff951af042e

*/

import (
        "crypto/tls"
        "crypto/x509"
        "encoding/json"
        "fmt"          
        "github.com/gorilla/mux"
        "github.com/olivere/elastic"
        "io/ioutil"                 
        "log"                       
        "net/http"                  
        "os"                        
        "strings"                   
        "context"                   
        "github.com/go-ini/ini"     
)                                   


var esCfg *ini.Section


var esCl *elastic.Client = nil

func attachToEs(esc *elastic.Client) (*elastic.Client) {
        if esc != nil {                                 
            return esc                                  
        }                                               

        certFile   := esCfg.Key("elasticsearch.cert").String()
        keyFile    := esCfg.Key("elasticsearch.keyfile").String()
        caFile     := esCfg.Key("elasticsearch.ca_certs").String()
        esHostname := esCfg.Key("elasticsearch.host").String()    
        esPort     := esCfg.Key("elasticsearch.port").String()    
        esProt     := "https"                                     
        if use_ssl, err := esCfg.Key("elasticsearch.use_ssl").Bool(); err != nil {
            if !use_ssl {                                                         
                esProt  = "http"                                                  
            }                                                                     
        }

        // Load client cert
        cert, err := tls.LoadX509KeyPair(certFile, keyFile)
        if err != nil {                                    
                log.Fatal(err)                             
        }                                                  

        // Load CA cert
        caCert, err := ioutil.ReadFile(caFile)
        if err != nil {                       
                log.Fatal(err)                
        }                                     
        caCertPool := x509.NewCertPool()      
        caCertPool.AppendCertsFromPEM(caCert) 

        // Setup HTTPS client
        tlsConfig := &tls.Config{
                Certificates: []tls.Certificate{cert},
                RootCAs:      caCertPool,             
                InsecureSkipVerify: true,             
        }                                             
        tlsConfig.BuildNameToCertificate()            
        transport := &http.Transport{TLSClientConfig: tlsConfig}
        client := &http.Client{Transport: transport}            

        esURL := fmt.Sprintf("%s://%s:%s", esProt, esHostname, esPort)
        resp, err := client.Get(esURL+"/_cat/indices")                
        if err != nil {                                               
            fmt.Println(err)                                          
        } else {                                                      
            contents, err := ioutil.ReadAll(resp.Body)                
            fmt.Printf("RESP: %s\n", string(contents))                
            if err != nil {                                           
                fmt.Println("Error:", err)                            
            }                                                         
        }                                                             

        esClient, err := elastic.NewClient(elastic.SetHttpClient(client),
                elastic.SetURL(esURL), elastic.SetScheme(esProt)) //, elastic.SetSniff(false))

        if err != nil {
                fmt.Println("Cannot get NewClient!!")
                log.Fatal(err)                       
        }                                            
        return esClient                              
}                                                    

var ANALYZER_EXE_FMT string = "/home/ravir/bin/sleep.sh %s %s"

func CreateRequest(w http.ResponseWriter, req *http.Request) {
        values := make(map[string]interface{}, 0)             
        _ = json.NewDecoder(req.Body).Decode(&values)         

        if _, ok := values["filters"]; !ok {
                values["filters"] = ""      
        }                                   
        exeStmt := fmt.Sprintf(ANALYZER_EXE_FMT, values["fcheck"], values["filters"])

        var attr = os.ProcAttr{}
        attr.Files = []*os.File{os.Stdin, os.Stdout, os.Stderr}
        args := strings.Split(exeStmt, " ")                    
        proc, err := os.StartProcess(args[0], args, &attr)     
        if err != nil {                                        
                fmt.Print(err)                                 
                json.NewEncoder(w).Encode("Failure")           
                return                                         
        }                                                      
        go proc.Wait()                                         
        // Fetch the requestId and return it                   

        json.NewEncoder(w).Encode("Success")
}                                           

func DeleteRequest(w http.ResponseWriter, req *http.Request) {
        //TODO                                                
}                                                             

func GetResults(w http.ResponseWriter, req *http.Request) {
        params := mux.Vars(req)                            
        reqId := params["reqId"]                           
        fmt.Println("ReqId:", reqId)                       
        esCl = attachToEs(esCl)                            
        esIndex := esCfg.Key("elasticsearch.idx_name").String()
        if esCl != nil {                                       
            ss := esCl.Search(esIndex)                         
            if ss != nil {                                     
                esQuery := elastic.NewMatchQuery("RequestId", reqId)
                rslt, _ := ss.Query(esQuery).FetchSource(true).Do(context.Background())
                srcs := []interface{}{}                                                
                for i:=0; i < int(rslt.TotalHits()); i++ {                             
                    srcs = append(srcs, rslt.Hits.Hits[i].Source)                      
                }                                                                      
                json.NewEncoder(w).Encode(srcs)                                        
                return
            }
        } else {
            fmt.Println("attach to ES Failed...")
        }
}

func main() {
        ccdir_ev := os.Getenv("CC_DIR")
        if ccdir_ev == "" {
            ccdir_ev = "/isan/lib"
        }
        cfg, err := ini.Load(ccdir_ev + "config.ini")
        if err != nil {
            fmt.Println("Cannot load Config.ini !!")
            os.Exit(1)
        }
        esCfg = cfg.Section("cc_result_writer")
        esCl = attachToEs(esCl)
        if esCl != nil {
            router := mux.NewRouter()
            router.HandleFunc("/cc/", CreateRequest).Methods("POST")
            router.HandleFunc("/cc/{reqId}", DeleteRequest).Methods("DELETE")
            router.HandleFunc("/cc/{reqId}/results", GetResults).Methods("GET")
            log.Fatal(http.ListenAndServe(":12345", router))
        }
}