Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

added nats sdk #371

Open
wants to merge 10 commits into
base: main
Choose a base branch
from
Open

added nats sdk #371

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
3678e1d
added nats sdk
vijeyash1 May 14, 2024
22f69ee
changed nats values
Nithunikzz May 14, 2024
8d1f910
nats sdk in ketall kubepug outdated rakees
an1l4 May 15, 2024
fc4d6b1
integrated the nats sdk
vijeyash1 May 23, 2024
8f416f4
ketall-test-change
an1l4 May 27, 2024
6dbb948
kubescore testing
Nithunikzz May 28, 2024
0e8a85c
kubescore testing
Nithunikzz May 28, 2024
2ef1e0f
outdated test case error fix
an1l4 May 28, 2024
ee7db41
kuberhealthy test fix
an1l4 May 28, 2024
2c3a1cd
trivy-testing
Nithunikzz May 28, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 17 additions & 42 deletions agent/kubviz/k8smetrics_agent.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,11 @@ import (

//"github.com/go-co-op/gocron"
"github.com/go-co-op/gocron"
"github.com/nats-io/nats.go"
"github.com/intelops/kubviz/constants"
"github.com/intelops/kubviz/pkg/nats/sdk"

"context"

"github.com/intelops/kubviz/pkg/mtlsnats"
"github.com/intelops/kubviz/pkg/opentelemetry"

"k8s.io/client-go/kubernetes"
Expand Down Expand Up @@ -56,8 +56,6 @@ const (
// nats token, natsurl, clustername
var (
ClusterName string = os.Getenv("CLUSTER_NAME")
token string = os.Getenv("NATS_TOKEN")
natsurl string = os.Getenv("NATS_ADDRESS")

//for local testing provide the location of kubeconfig
cluster_conf_loc string = os.Getenv("CONFIG_LOCATION")
Expand All @@ -77,36 +75,13 @@ func main() {
clientset *kubernetes.Clientset
)

var mtlsConfig mtlsnats.MtlsConfig
var nc *nats.Conn

if mtlsConfig.IsEnabled {
tlsConfig, err := mtlsnats.GetTlsConfig()
if err != nil {
log.Println("error while getting tls config ", err)
time.Sleep(time.Minute * 30)
} else {
nc, err = nats.Connect(
natsurl,
nats.Name("K8s Metrics"),
nats.Token(token),
nats.Secure(tlsConfig),
)
if err != nil {
log.Println("error while connecting with mtls ", err)
}
}
natsCli, err := sdk.NewNATSClient()

if err != nil {
log.Fatalf("error occured while creating nats client %v", err.Error())
}

if nc == nil {
nc, err = nats.Connect(natsurl, nats.Name("K8s Metrics"), nats.Token(token))
events.CheckErr(err)
}
js, err := nc.JetStream()
events.CheckErr(err)
err = events.CreateStream(js)
events.CheckErr(err)
natsCli.CreateStream(constants.StreamName)
if env != Production {
config, err = clientcmd.BuildConfigFromFlags("", cluster_conf_loc)
if err != nil {
Expand All @@ -131,32 +106,32 @@ func main() {
}
}()

go events.PublishMetrics(clientset, js, clusterMetricsChan)
go events.PublishMetrics(clientset, natsCli, clusterMetricsChan)
if cfg.KuberHealthyEnable {
go kuberhealthy.StartKuberHealthy(js)
go kuberhealthy.StartKuberHealthy(natsCli)
}
go server.StartServer()
collectAndPublishMetrics := func() {
err := outdated.OutDatedImages(config, js)
err := outdated.OutDatedImages(config, natsCli)
events.LogErr(err)
err = kubepreupgrade.KubePreUpgradeDetector(config, js)
err = kubepreupgrade.KubePreUpgradeDetector(config, natsCli)
events.LogErr(err)
err = ketall.GetAllResources(config, js)
err = ketall.GetAllResources(config, natsCli)
events.LogErr(err)
err = rakkess.RakeesOutput(config, js)
err = rakkess.RakeesOutput(config, natsCli)
events.LogErr(err)
err = trivy.RunTrivySbomScan(config, js)
err = trivy.RunTrivySbomScan(config, natsCli)
events.LogErr(err)
err = trivy.RunTrivyImageScans(config, js)
err = trivy.RunTrivyImageScans(config, natsCli)
events.LogErr(err)
err = trivy.RunTrivyK8sClusterScan(js)
err = trivy.RunTrivyK8sClusterScan(natsCli)
events.LogErr(err)
err = kubescore.RunKubeScore(clientset, js)
err = kubescore.RunKubeScore(clientset, natsCli)
events.LogErr(err)
}
collectAndPublishMetrics()
if cfg.SchedulerEnable { // Assuming "cfg.Schedule" is a boolean indicating whether to schedule or not.
scheduler := scheduler.InitScheduler(config, js, *cfg, clientset)
scheduler := scheduler.InitScheduler(config, natsCli, *cfg, clientset)

// Start the scheduler
scheduler.Start()
Expand Down
17 changes: 9 additions & 8 deletions agent/kubviz/plugins/events/event_metrics_utils.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (

"github.com/intelops/kubviz/constants"
"github.com/intelops/kubviz/model"
"github.com/intelops/kubviz/pkg/nats/sdk"
"github.com/intelops/kubviz/pkg/opentelemetry"
"github.com/nats-io/nats.go"
"go.opentelemetry.io/otel"
Expand All @@ -29,19 +30,19 @@ var ClusterName string = os.Getenv("CLUSTER_NAME")

// publishMetrics publishes stream of events
// with subject "METRICS.created"
func PublishMetrics(clientset *kubernetes.Clientset, js nats.JetStreamContext, errCh chan error) {
func PublishMetrics(clientset *kubernetes.Clientset, natsCli *sdk.NATSClient, errCh chan error) {

ctx := context.Background()
tracer := otel.Tracer("kubviz-publish-metrics")
_, span := tracer.Start(opentelemetry.BuildContext(ctx), "publishMetrics")
span.SetAttributes(attribute.String("kubviz-agent", "publish-metrics"))
defer span.End()

watchK8sEvents(clientset, js)
watchK8sEvents(clientset, natsCli)
errCh <- nil
}

func publishK8sMetrics(id string, mtype string, mdata *v1.Event, js nats.JetStreamContext, imageName string) (bool, error) {
func publishK8sMetrics(id string, mtype string, mdata *v1.Event, natsCli *sdk.NATSClient, imageName string) (bool, error) {

ctx := context.Background()
tracer := otel.Tracer("kubviz-publish-k8smetrics")
Expand All @@ -57,7 +58,7 @@ func publishK8sMetrics(id string, mtype string, mdata *v1.Event, js nats.JetStre
ImageName: imageName,
}
metricsJson, _ := json.Marshal(metrics)
_, err := js.Publish(constants.EventSubject, metricsJson)
err := natsCli.Publish(constants.EventSubject, metricsJson)
if err != nil {
return true, err
}
Expand Down Expand Up @@ -164,7 +165,7 @@ func LogErr(err error) {
log.Println(err)
}
}
func watchK8sEvents(clientset *kubernetes.Clientset, js nats.JetStreamContext) {
func watchK8sEvents(clientset *kubernetes.Clientset, natsCli *sdk.NATSClient) {

ctx := context.Background()
tracer := otel.Tracer("kubviz-watch-k8sevents")
Expand All @@ -191,7 +192,7 @@ func watchK8sEvents(clientset *kubernetes.Clientset, js nats.JetStreamContext) {
return
}
for _, image := range images {
publishK8sMetrics(string(event.ObjectMeta.UID), "ADD", event, js, image)
publishK8sMetrics(string(event.ObjectMeta.UID), "ADD", event, natsCli, image)
}
},
DeleteFunc: func(obj interface{}) {
Expand All @@ -202,7 +203,7 @@ func watchK8sEvents(clientset *kubernetes.Clientset, js nats.JetStreamContext) {
return
}
for _, image := range images {
publishK8sMetrics(string(event.ObjectMeta.UID), "DELETE", event, js, image)
publishK8sMetrics(string(event.ObjectMeta.UID), "DELETE", event, natsCli, image)
}
},
UpdateFunc: func(oldObj, newObj interface{}) {
Expand All @@ -213,7 +214,7 @@ func watchK8sEvents(clientset *kubernetes.Clientset, js nats.JetStreamContext) {
return
}
for _, image := range images {
publishK8sMetrics(string(event.ObjectMeta.UID), "UPDATE", event, js, image)
publishK8sMetrics(string(event.ObjectMeta.UID), "UPDATE", event, natsCli, image)
}
},
},
Expand Down
10 changes: 5 additions & 5 deletions agent/kubviz/plugins/ketall/ketall.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ import (
"go.opentelemetry.io/otel/attribute"

"github.com/intelops/kubviz/model"
"github.com/nats-io/nats.go"
"github.com/intelops/kubviz/pkg/nats/sdk"
log "github.com/sirupsen/logrus"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/discovery"
Expand All @@ -22,19 +22,19 @@ import (

var ClusterName string = os.Getenv("CLUSTER_NAME")

func PublishAllResources(result model.Resource, js nats.JetStreamContext) error {
func PublishAllResources(result model.Resource, natsCli *sdk.NATSClient) error {
metrics := result
metrics.ClusterName = ClusterName
metricsJson, _ := json.Marshal(metrics)
_, err := js.Publish(constants.EventSubject_getall_resource, metricsJson)
err := natsCli.Publish(constants.EventSubject_getall_resource, metricsJson)
if err != nil {
return err
}
log.Printf("Metrics with resource %s in the %s namespace has been published", result.Resource, result.Namespace)
return nil
}

func GetAllResources(config *rest.Config, js nats.JetStreamContext) error {
func GetAllResources(config *rest.Config, natsCli *sdk.NATSClient) error {

ctx := context.Background()
tracer := otel.Tracer("ketall")
Expand Down Expand Up @@ -88,7 +88,7 @@ func GetAllResources(config *rest.Config, js nats.JetStreamContext) error {
}

}
err := PublishAllResources(resource, js)
err := PublishAllResources(resource, natsCli)
if err != nil {
return err
}
Expand Down
Loading
Loading