Add tool list_unstructured_data_pipelines_for_user#254
Conversation
gshikhar2021
commented
Jun 26, 2026
gshikhar2021
commented
- this tool will list all the CR from kubernetes cluster using service account role and access using incluster config
- list all the databases from snowflake using user's OAuth token
- this tool will list all the CR from kubernetes cluster using service account role and access - list all the databases from snowflake
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces a new Model Context Protocol (MCP) server component to the controller. The primary goal is to provide AI assistants with the ability to query and list both Kubernetes-based data pipelines and Snowflake databases. The changes include the necessary client-side logic for both platforms, secure token handling within the context, and the full suite of Kubernetes manifests required to deploy the server within the cluster. Highlights
Ignored Files
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces an MCP server for the unstructured data controller, adding Dockerfile configurations, deployment manifests, and tools to list pipelines and Snowflake databases. Key feedback includes removing a hardcoded 'SHIKHAR%' pattern from the Snowflake database query, simplifying redundant connection pool configurations on short-lived database connections, allowing the pipeline listing tool to return partial results upon partial failures, and adding a local kubeconfig fallback to the Kubernetes client to improve the local development experience.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
|
|
||
| // Run SHOW DATABASES query | ||
| // This will return only databases the authenticated user has access to (RBAC enforced by Snowflake) | ||
| rows, err := db.QueryContext(ctx, "SHOW DATABASES LIKE 'SHIKHAR%';") |
There was a problem hiding this comment.
The query SHOW DATABASES LIKE 'SHIKHAR%'; contains a hardcoded pattern 'SHIKHAR%' which appears to be a leftover from development/testing. This will prevent users from seeing any databases that do not match this prefix. It should be removed to list all databases the user has access to.
| rows, err := db.QueryContext(ctx, "SHOW DATABASES LIKE 'SHIKHAR%';") | |
| rows, err := db.QueryContext(ctx, "SHOW DATABASES;") |
| // Configure connection pool for security | ||
| // MaxOpenConns=1: Single connection per request (we open/close immediately) | ||
| // MaxIdleConns=0: Don't keep connections idle (reduces token exposure time) | ||
| // ConnMaxLifetime=5min: Close connections after 5 minutes (limits token lifetime) | ||
| db.SetMaxOpenConns(1) | ||
| db.SetMaxIdleConns(0) | ||
| db.SetConnMaxLifetime(5 * time.Minute) |
There was a problem hiding this comment.
Since the db connection pool is opened and immediately closed (via defer db.Close()) within the lifecycle of a single ShowDatabases call, configuring connection pool limits like SetMaxIdleConns(0) and SetConnMaxLifetime(5 * time.Minute) is redundant and has no effect. If you intend to keep the connection pool open across multiple requests, you should manage the sql.DB lifecycle outside of this function. Otherwise, these configuration lines can be simplified or removed.
| // List Kubernetes UnstructuredDataPipeline CRs | ||
| var pipelines []k8sclient.PipelineInfo | ||
| if k8sClient != nil { | ||
| var err error | ||
| pipelines, err = k8sClient.ListPipelines(ctx) | ||
| if err != nil { | ||
| return &mcp.CallToolResult{ | ||
| Content: []mcp.Content{&mcp.TextContent{ | ||
| Text: fmt.Sprintf("Error listing pipelines: %v", err), | ||
| }}, | ||
| IsError: true, | ||
| }, nil, nil | ||
| } | ||
| } | ||
|
|
||
| // Query Snowflake databases using the user's OAuth token | ||
| databases, err := snowflake.ShowDatabases(ctx, oauthToken) | ||
| if err != nil { | ||
| return &mcp.CallToolResult{ | ||
| Content: []mcp.Content{&mcp.TextContent{ | ||
| Text: fmt.Sprintf("Error querying Snowflake: %v", err), | ||
| }}, | ||
| IsError: true, | ||
| }, nil, nil | ||
| } |
There was a problem hiding this comment.
The tool currently fails completely if either the Kubernetes client or the Snowflake client encounters an error. Consider making the tool more resilient by returning partial results (e.g., returning the successfully fetched pipelines even if Snowflake queries fail, or vice versa) along with an error/warning message in the response content, rather than failing the entire request.
| func NewInClusterClient() (*Client, error) { | ||
| // rest.InClusterConfig() reads from: | ||
| // 1. /var/run/secrets/kubernetes.io/serviceaccount/token (for auth) | ||
| // 2. /var/run/secrets/kubernetes.io/serviceaccount/ca.crt (for TLS) | ||
| // 3. KUBERNETES_SERVICE_HOST and KUBERNETES_SERVICE_PORT env vars | ||
| config, err := rest.InClusterConfig() | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to get in-cluster config: %w", err) | ||
| } |
There was a problem hiding this comment.
The NewInClusterClient function currently only supports in-cluster configuration. If the server is run locally for development or debugging, it will fail to start and exit. Consider falling back to loading the local kubeconfig (using clientcmd.NewNonInteractiveDeferredLoadingClientConfig) when rest.InClusterConfig() returns rest.ErrNotInCluster. This significantly improves the local development experience.
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces a Model Context Protocol (MCP) server deployment, including a Dockerfile, Makefile targets, Kubernetes manifests, and tools for health checks and listing pipelines. Key feedback focuses on removing a hardcoded testing query in the Snowflake client, dynamically detecting the Kubernetes namespace instead of hardcoding it, utilizing config.GetConfig() to support seamless local development, and replacing a hardcoded condition string with an existing API constant.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
|
|
||
| // Run SHOW DATABASES query | ||
| // This will return only databases the authenticated user has access to (RBAC enforced by Snowflake) | ||
| rows, err := db.QueryContext(ctx, "SHOW DATABASES LIKE 'SHIKHAR%';") |
There was a problem hiding this comment.
The query SHOW DATABASES LIKE 'SHIKHAR%'; is hardcoded to only list databases starting with 'SHIKHAR%'. This appears to be a leftover from local testing and will prevent other users from listing their databases. It should be changed to a generic SHOW DATABASES; query to list all databases the authenticated user has access to.
| rows, err := db.QueryContext(ctx, "SHOW DATABASES LIKE 'SHIKHAR%';") | |
| rows, err := db.QueryContext(ctx, "SHOW DATABASES;") |
| // Client wraps a Kubernetes client for accessing cluster resources | ||
| type Client struct { | ||
| client client.Client | ||
| } |
There was a problem hiding this comment.
Add a namespace field to the Client struct so that we can dynamically list pipelines in the namespace where the MCP server is deployed, rather than relying on a hardcoded namespace.
| // Client wraps a Kubernetes client for accessing cluster resources | |
| type Client struct { | |
| client client.Client | |
| } | |
| // Client wraps a Kubernetes client for accessing cluster resources | |
| type Client struct { | |
| client client.Client | |
| namespace string | |
| } |
| func NewInClusterClient() (*Client, error) { | ||
| // rest.InClusterConfig() reads from: | ||
| // 1. /var/run/secrets/kubernetes.io/serviceaccount/token (for auth) | ||
| // 2. /var/run/secrets/kubernetes.io/serviceaccount/ca.crt (for TLS) | ||
| // 3. KUBERNETES_SERVICE_HOST and KUBERNETES_SERVICE_PORT env vars | ||
| config, err := rest.InClusterConfig() | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to get in-cluster config: %w", err) | ||
| } | ||
|
|
||
| // Create a controller-runtime client with our scheme (includes CRDs) | ||
| k8sClient, err := client.New(config, client.Options{Scheme: scheme}) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to create kubernetes client: %w", err) | ||
| } | ||
|
|
||
| return &Client{ | ||
| client: k8sClient, | ||
| }, nil | ||
| } |
There was a problem hiding this comment.
Update the client constructor to use config.GetConfig() for automatic local/in-cluster configuration loading, and dynamically read the current namespace from the service account mount path /var/run/secrets/kubernetes.io/serviceaccount/namespace with a fallback to the default namespace.
func NewInClusterClient() (*Client, error) {
// config.GetConfig() automatically handles both in-cluster config
// and local kubeconfig fallback, making local development seamless.
cfg, err := config.GetConfig()
if err != nil {
return nil, fmt.Errorf("failed to get kubernetes config: %w", err)
}
// Create a controller-runtime client with our scheme (includes CRDs)
k8sClient, err := client.New(cfg, client.Options{Scheme: scheme})
if err != nil {
return nil, fmt.Errorf("failed to create kubernetes client: %w", err)
}
// Read the namespace from the service account directory, falling back to default
ns := "unstructured-controller-namespace"
if data, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace"); err == nil {
ns = strings.TrimSpace(string(data))
}
return &Client{
client: k8sClient,
namespace: ns,
}, nil
}| func (c *Client) ListPipelines(ctx context.Context) ([]PipelineInfo, error) { | ||
| pipelineList := &operatorv1alpha1.UnstructuredDataPipelineList{} | ||
|
|
||
| // List all pipelines in the unstructured-controller-namespace | ||
| // This will use the ServiceAccount's RBAC permissions | ||
| err := c.client.List(ctx, pipelineList, &client.ListOptions{ | ||
| Namespace: "unstructured-controller-namespace", | ||
| }) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to list pipelines: %w", err) | ||
| } |
There was a problem hiding this comment.
Use the dynamically detected namespace instead of the hardcoded namespace.
| func (c *Client) ListPipelines(ctx context.Context) ([]PipelineInfo, error) { | |
| pipelineList := &operatorv1alpha1.UnstructuredDataPipelineList{} | |
| // List all pipelines in the unstructured-controller-namespace | |
| // This will use the ServiceAccount's RBAC permissions | |
| err := c.client.List(ctx, pipelineList, &client.ListOptions{ | |
| Namespace: "unstructured-controller-namespace", | |
| }) | |
| if err != nil { | |
| return nil, fmt.Errorf("failed to list pipelines: %w", err) | |
| } | |
| func (c *Client) ListPipelines(ctx context.Context) ([]PipelineInfo, error) { | |
| pipelineList := &operatorv1alpha1.UnstructuredDataPipelineList{} | |
| // List all pipelines in the detected namespace | |
| // This will use the ServiceAccount's RBAC permissions | |
| err := c.client.List(ctx, pipelineList, &client.ListOptions{ | |
| Namespace: c.namespace, | |
| }) | |
| if err != nil { | |
| return nil, fmt.Errorf("failed to list pipelines: %w", err) | |
| } |
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| operatorv1alpha1 "github.com/redhat-data-and-ai/unstructured-data-controller/api/v1alpha1" | ||
| "k8s.io/apimachinery/pkg/runtime" | ||
| utilruntime "k8s.io/apimachinery/pkg/util/runtime" | ||
| clientgoscheme "k8s.io/client-go/kubernetes/scheme" | ||
| "k8s.io/client-go/rest" | ||
| "sigs.k8s.io/controller-runtime/pkg/client" | ||
| ) |
There was a problem hiding this comment.
To support seamless local development and testing, we can use sigs.k8s.io/controller-runtime/pkg/client/config's GetConfig() which automatically falls back to the local kubeconfig when run outside of a Kubernetes cluster. Let's update the imports to include the config package and other standard packages needed for dynamic namespace detection.
import (
"context"
"fmt"
"os"
"strings"
operatorv1alpha1 "github.com/redhat-data-and-ai/unstructured-data-controller/api/v1alpha1"
"k8s.io/apimachinery/pkg/runtime"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
clientgoscheme "k8s.io/client-go/kubernetes/scheme"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/client/config"
)| for _, condition := range pipeline.Status.Conditions { | ||
| if condition.Type == "UnstructuredDataPipelineReady" { | ||
| info.Status = string(condition.Status) |
There was a problem hiding this comment.
The condition type "UnstructuredDataPipelineReady" is hardcoded here, but there is already a constant UnstructuredDataPipelineCondition defined in api/v1alpha1/unstructureddatapipeline_types.go for this exact purpose. Using the defined constant improves maintainability and prevents potential typos.
| for _, condition := range pipeline.Status.Conditions { | |
| if condition.Type == "UnstructuredDataPipelineReady" { | |
| info.Status = string(condition.Status) | |
| for _, condition := range pipeline.Status.Conditions { | |
| if condition.Type == operatorv1alpha1.UnstructuredDataPipelineCondition { | |
| info.Status = string(condition.Status) |
2 participants
